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

fix dc connector get catalogs metadata from remote data center too frequently.

This commit is contained in:
giteezhangjingfang 2020-07-06 19:33:26 +08:00 committed by zhangjingfang
parent 56ac7443b2
commit 4d18bf9aca
14 changed files with 96 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/DataCenterConfig.java
View File

@ -153,7 +153,7 @@ public class DataCenterConfig
* @return DataCenterConfig object.
*/
@Config("dc.http-request-readTimeout")
@ConfigDescription("http request read timeout, default value is 2min")
@ConfigDescription("http request read timeout, default value is 30s")
public DataCenterConfig setHttpRequestReadTimeout(Duration httpRequestReadTimeout)
{
this.httpRequestReadTimeout = httpRequestReadTimeout;
@ -177,7 +177,7 @@ public class DataCenterConfig
* @return DataCenterConfig object.
*/
@Config("dc.http-request-connectTimeout")
@ConfigDescription("http request connect timeout, default value is 1min")
@ConfigDescription("http request connect timeout, default value is 30s")
public DataCenterConfig setHttpRequestConnectTimeout(Duration httpRequestConnectTimeout)
{
this.httpRequestConnectTimeout = httpRequestConnectTimeout;
@ -778,7 +778,7 @@ public class DataCenterConfig
* @return DataCenterConfig object
*/
@Config("dc.http-compression")
@ConfigDescription("whether use gzip compress response body, default value is true")
@ConfigDescription("whether use gzip compress response body, default value is false")
public DataCenterConfig setCompressionEnabled(boolean isCompressionEnabledParameter)
{
this.isCompressionEnabled = isCompressionEnabledParameter;
@ -797,7 +797,7 @@ public class DataCenterConfig
* @return DataCenterConfig object
*/
@Config("dc.httpclient.maximum.idle.connections")
@ConfigDescription("whether use gzip compress response body, default value is true")
@ConfigDescription("http client maximum idle connections")
public DataCenterConfig setMaxIdleConnections(int maxIdleConnectionsParameter)
{
this.maxIdleConnections = maxIdleConnectionsParameter;

6
hetu-docs/en/connector/datacenter.md
View File

@ -28,9 +28,9 @@ The following table lists all properties supported by the datacenter connector.
| `dc.application.name.prefix` | Prefix to append to any specified ApplicationName client info property, which is used to Set source name for the openLooKeng query. If neither this property nor ApplicationName are set, the source for the query will be hetu-dc | `hetu-dc` |
| `dc.extra.credentials` | Extra credentials for connecting to external services. The extraCredentials is a list of key-value pairs. Example: foo:bar;abc:xyz will create credentials abc=xyz and foo=bar | |
| `dc.http-client-timeout` | Time until the client keeps retrying to fetch the data, default value is 10min | `10.00m` |
| `dc.http-compression` | Whether use gzip compress response body, default value is true | `false` |
| `dc.http-request-connectTimeout` | http request connect timeout, default value is 1min | `30.00s` |
| `dc.http-request-readTimeout` | http request read timeout, default value is 2min | `30.00s` |
| `dc.http-compression` | Whether use gzip compress response body, default value is false | `false` |
| `dc.http-request-connectTimeout` | http request connect timeout, default value is 30s | `30.00s` |
| `dc.http-request-readTimeout` | http request read timeout, default value is 30s | `30.00s` |
| `dc.httpclient.maximum.idle.connections` | Maximum idle connections to be kept open in the http client | `20` |
| `dc.httpproxy` | HTTP proxy host and port. Example: localhost:8888 | |
| `dc.kerberos.config.path` | Kerberos configuration file | |

6
hetu-docs/zh/connector/datacenter.md
View File

@ -25,9 +25,9 @@ connection-password=password
| `dc.application.name.prefix`| 添加到任何指定的ApplicationName客户端信息属性的前缀，该前缀用于设置openLooKeng查询的源名称。如果没有设置此属性或ApplicationName，则查询的源将是hetu-dc。| `hetu-dc`|
| `dc.extra.credentials`| 连接外部服务的额外凭证。ExtraCredentials是一个键值对列表。示例：**foo:bar;abc:xyz**将创建凭证**abc=xyz**和**foo=bar**。| |
| `dc.http-client-timeout`| 客户端持续重试取数据的时间，默认值为10分钟| `10.00m`|
| `dc.http-compression`| 是否使用gzip压缩响应体，默认值为true| `false`|
| `dc.http-request-connectTimeout`| HTTP请求连接超时，默认值为1分钟| `30.00s`|
| `dc.http-request-readTimeout`| HTTP请求读取超时，默认为2分钟| `30.00s`|
| `dc.http-compression`| 是否使用gzip压缩响应体，默认值为false| `false`|
| `dc.http-request-connectTimeout`| HTTP请求连接超时，默认值为30秒| `30.00s`|
| `dc.http-request-readTimeout`| HTTP请求读取超时，默认为30秒| `30.00s`|
| `dc.httpclient.maximum.idle.connections`| HTTP客户端保持打开的最大空闲连接| `20`|
| `dc.httpproxy`| HTTP代理主机和端口。示例：**localhost:8888**| |
| `dc.kerberos.config.path`| Kerberos配置文件| |

40
presto-main/src/main/java/io/prestosql/connector/DataCenterUtility.java
View File

@ -16,13 +16,17 @@ package io.prestosql.connector;
import io.airlift.log.Logger;
import io.prestosql.Session;
import io.prestosql.execution.TaskSource;
import io.prestosql.metadata.Metadata;
import io.prestosql.metadata.MetadataManager;
import io.prestosql.spi.PrestoTransportException;
import io.prestosql.sql.tree.Identifier;
import io.prestosql.sql.tree.Use;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
public class DataCenterUtility
{
@ -90,6 +94,42 @@ public class DataCenterUtility
}
}
/**
* Utility method to load all DC catalog from updateTask.
*
* @param metadata session
* @param sources task source
*/
public static void loadDCCatalogForUpdateTask(Metadata metadata, List<TaskSource> sources)
{
if (metadata instanceof MetadataManager) {
// try to load dc catalog
DataCenterConnectorManager dataCenterConnectorManager = ((MetadataManager) metadata).getDataCenterConnectorManager();
Set<String> catalogNames = new HashSet<>();
sources.stream().forEach(source -> {
source.getSplits().stream()
.forEach(split ->
{
String catalogName = split.getSplit().getCatalogName().getCatalogName();
if (catalogName.contains(".")
&& dataCenterConnectorManager.isDCCatalog(catalogName.substring(0, catalogName.indexOf(".")))) {
catalogNames.add(catalogName);
}
});
});
catalogNames.stream()
.forEach(catalogName -> {
dataCenterConnectorManager.loadDCCatalog(catalogName);
});
}
}
/**
* Utility method to check whether a catalog is cross-dc catalog.
*
* @param metadata session
* @param catalogName catalog name
*/
public static boolean isDCCatalog(Metadata metadata, String catalogName)
{
MetadataManager metadataManager = (MetadataManager) metadata;

13
presto-main/src/main/java/io/prestosql/execution/SqlTask.java
View File

@ -31,6 +31,7 @@ import io.prestosql.execution.buffer.OutputBuffer;
import io.prestosql.execution.buffer.OutputBuffers;
import io.prestosql.execution.buffer.OutputBuffers.OutputBufferId;
import io.prestosql.memory.QueryContext;
import io.prestosql.metadata.Metadata;
import io.prestosql.operator.PipelineContext;
import io.prestosql.operator.PipelineStatus;
import io.prestosql.operator.TaskContext;
@ -58,6 +59,7 @@ import static com.google.common.util.concurrent.Futures.immediateFuture;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static io.airlift.units.DataSize.Unit.BYTE;
import static io.airlift.units.DataSize.succinctBytes;
import static io.prestosql.connector.DataCenterUtility.loadDCCatalogForUpdateTask;
import static io.prestosql.execution.TaskState.ABORTED;
import static io.prestosql.execution.TaskState.FAILED;
import static io.prestosql.util.Failures.toFailures;
@ -83,6 +85,7 @@ public class SqlTask
private final AtomicReference<TaskHolder> taskHolderReference = new AtomicReference<>(new TaskHolder());
private final AtomicBoolean needsPlan = new AtomicBoolean(true);
private final Metadata metadata;
public static SqlTask createSqlTask(
TaskId taskId,
@ -93,9 +96,10 @@ public class SqlTask
ExecutorService taskNotificationExecutor,
Function<SqlTask, ?> onDone,
DataSize maxBufferSize,
CounterStat failedTasks)
CounterStat failedTasks,
Metadata metadata)
{
SqlTask sqlTask = new SqlTask(taskId, location, nodeId, queryContext, sqlTaskExecutionFactory, taskNotificationExecutor, maxBufferSize);
SqlTask sqlTask = new SqlTask(taskId, location, nodeId, queryContext, sqlTaskExecutionFactory, taskNotificationExecutor, maxBufferSize, metadata);
sqlTask.initialize(onDone, failedTasks);
return sqlTask;
}
@ -107,7 +111,8 @@ public class SqlTask
QueryContext queryContext,
SqlTaskExecutionFactory sqlTaskExecutionFactory,
ExecutorService taskNotificationExecutor,
DataSize maxBufferSize)
DataSize maxBufferSize,
Metadata metadata)
{
this.taskId = requireNonNull(taskId, "taskId is null");
this.taskInstanceId = UUID.randomUUID().toString();
@ -115,6 +120,7 @@ public class SqlTask
this.nodeId = requireNonNull(nodeId, "nodeId is null");
this.queryContext = requireNonNull(queryContext, "queryContext is null");
this.sqlTaskExecutionFactory = requireNonNull(sqlTaskExecutionFactory, "sqlTaskExecutionFactory is null");
this.metadata = requireNonNull(metadata, "requireNonNull is null");
requireNonNull(taskNotificationExecutor, "taskNotificationExecutor is null");
requireNonNull(maxBufferSize, "maxBufferSize is null");
@ -380,6 +386,7 @@ public class SqlTask
taskExecution = taskHolder.getTaskExecution();
if (taskExecution == null) {
checkState(fragment.isPresent(), "fragment must be present");
loadDCCatalogForUpdateTask(metadata, sources);
taskExecution = sqlTaskExecutionFactory.create(session, queryContext, taskStateMachine, outputBuffer, fragment.get(), sources, totalPartitions);
taskHolderReference.compareAndSet(taskHolder, new TaskHolder(taskExecution));
needsPlan.set(false);

7
presto-main/src/main/java/io/prestosql/execution/SqlTaskManager.java
View File

@ -39,6 +39,7 @@ import io.prestosql.memory.MemoryPoolAssignment;
import io.prestosql.memory.MemoryPoolAssignmentsRequest;
import io.prestosql.memory.NodeMemoryConfig;
import io.prestosql.memory.QueryContext;
import io.prestosql.metadata.Metadata;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.QueryId;
import io.prestosql.spiller.LocalSpillManager;
@ -121,7 +122,8 @@ public class SqlTaskManager
NodeMemoryConfig nodeMemoryConfig,
LocalSpillManager localSpillManager,
NodeSpillConfig nodeSpillConfig,
GcMonitor gcMonitor)
GcMonitor gcMonitor,
Metadata metadata)
{
requireNonNull(nodeInfo, "nodeInfo is null");
requireNonNull(config, "config is null");
@ -159,7 +161,8 @@ public class SqlTaskManager
return null;
},
maxBufferSize,
failedTasks)));
failedTasks,
metadata)));
}
private QueryContext createQueryContext(

12
presto-main/src/main/java/io/prestosql/split/PageSourceManager.java
View File

@ -13,10 +13,8 @@
*/
package io.prestosql.split;
import com.google.inject.Inject;
import io.prestosql.Session;
import io.prestosql.connector.CatalogName;
import io.prestosql.metadata.Metadata;
import io.prestosql.metadata.Split;
import io.prestosql.metadata.TableHandle;
import io.prestosql.spi.connector.ColumnHandle;
@ -34,20 +32,12 @@ import java.util.function.Supplier;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static io.prestosql.SystemSessionProperties.getDynamicFilteringWaitTime;
import static io.prestosql.connector.DataCenterUtility.loadDCCatalogForQueryFlow;
import static java.util.Objects.requireNonNull;
public class PageSourceManager
implements PageSourceProvider
{
private final ConcurrentMap<CatalogName, ConnectorPageSourceProvider> pageSourceProviders = new ConcurrentHashMap<>();
private final Metadata metadata;
@Inject
public PageSourceManager(Metadata metadata)
{
this.metadata = metadata;
}
public void addConnectorPageSourceProvider(CatalogName catalogName, ConnectorPageSourceProvider pageSourceProvider)
{
@ -68,8 +58,6 @@ public class PageSourceManager
checkArgument(split.getCatalogName().equals(table.getCatalogName()), "mismatched split and table");
CatalogName catalogName = split.getCatalogName();
loadDCCatalogForQueryFlow(session, metadata, catalogName.getCatalogName());
ConnectorPageSourceProvider provider = getPageSourceProvider(catalogName);
Map<ColumnHandle, DynamicFilter> constraint = null;

2
presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java
View File

@ -333,7 +333,7 @@ public class LocalQueryRunner
this.costCalculator = new CostCalculatorUsingExchanges(taskCountEstimator);
this.estimatedExchangesCostCalculator = new CostCalculatorWithEstimatedExchanges(costCalculator, taskCountEstimator);
this.accessControl = new TestingAccessControlManager(transactionManager);
this.pageSourceManager = new PageSourceManager(metadata);
this.pageSourceManager = new PageSourceManager();
this.pageFunctionCompiler = new PageFunctionCompiler(metadata, 0);
this.expressionCompiler = new ExpressionCompiler(metadata, pageFunctionCompiler);

1
presto-main/src/test/java/io/prestosql/catalog/TestDynamicCatalogRunner.java
View File

@ -108,7 +108,6 @@ public class TestDynamicCatalogRunner
server.installPlugin(new HetuFileSystemClientPlugin());
server.getInstance(Key.get(FileSystemClientManager.class)).loadFactoryConfigs();
}
server.getInstance(Key.get(CatalogStoreUtil.class));
}
private void prepareDirectory(String directory)

23
presto-main/src/test/java/io/prestosql/execution/EmptyMockMetadata.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.prestosql.execution;
import io.prestosql.metadata.AbstractMockMetadata;
public class EmptyMockMetadata
extends AbstractMockMetadata
{
}

2
presto-main/src/test/java/io/prestosql/execution/TaskTestUtils.java
View File

@ -108,7 +108,7 @@ public final class TaskTestUtils
{
Metadata metadata = createTestMetadataManager();
PageSourceManager pageSourceManager = new PageSourceManager(metadata);
PageSourceManager pageSourceManager = new PageSourceManager();
pageSourceManager.addConnectorPageSourceProvider(CONNECTOR_ID, new TestingPageSourceProvider());
// we don't start the finalizer so nothing will be collected, which is ok for a test

3
presto-main/src/test/java/io/prestosql/execution/TestMemoryRevokingScheduler.java
View File

@ -306,6 +306,7 @@ public class TestMemoryRevokingScheduler
executor,
Functions.identity(),
new DataSize(32, MEGABYTE),
new CounterStat());
new CounterStat(),
new EmptyMockMetadata());
}
}

3
presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java
View File

@ -322,6 +322,7 @@ public class TestSqlTask
taskNotificationExecutor,
Functions.identity(),
new DataSize(32, MEGABYTE),
new CounterStat());
new CounterStat(),
new EmptyMockMetadata());
}
}

3
presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java
View File

@ -245,7 +245,8 @@ public class TestSqlTaskManager
new NodeMemoryConfig(),
localSpillManager,
new NodeSpillConfig(),
new TestingGcMonitor());
new TestingGcMonitor(),
new EmptyMockMetadata());
}
private TaskInfo createTask(SqlTaskManager sqlTaskManager, TaskId taskId, ImmutableSet<ScheduledSplit> splits, OutputBuffers outputBuffers)