diff --git a/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/DataCenterConfig.java b/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/DataCenterConfig.java index 175271ecf..1d99bda12 100644 --- a/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/DataCenterConfig.java +++ b/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/DataCenterConfig.java @@ -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; diff --git a/hetu-docs/en/connector/datacenter.md b/hetu-docs/en/connector/datacenter.md index b59e8210d..d530611cc 100644 --- a/hetu-docs/en/connector/datacenter.md +++ b/hetu-docs/en/connector/datacenter.md @@ -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 | | diff --git a/hetu-docs/zh/connector/datacenter.md b/hetu-docs/zh/connector/datacenter.md index 3a560946a..039d07db8 100644 --- a/hetu-docs/zh/connector/datacenter.md +++ b/hetu-docs/zh/connector/datacenter.md @@ -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配置文件| | diff --git a/presto-main/src/main/java/io/prestosql/connector/DataCenterUtility.java b/presto-main/src/main/java/io/prestosql/connector/DataCenterUtility.java index 403e44421..6ef2fc0a4 100644 --- a/presto-main/src/main/java/io/prestosql/connector/DataCenterUtility.java +++ b/presto-main/src/main/java/io/prestosql/connector/DataCenterUtility.java @@ -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 sources) + { + if (metadata instanceof MetadataManager) { + // try to load dc catalog + DataCenterConnectorManager dataCenterConnectorManager = ((MetadataManager) metadata).getDataCenterConnectorManager(); + Set 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; diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlTask.java b/presto-main/src/main/java/io/prestosql/execution/SqlTask.java index 90d540bd3..44993b28b 100644 --- a/presto-main/src/main/java/io/prestosql/execution/SqlTask.java +++ b/presto-main/src/main/java/io/prestosql/execution/SqlTask.java @@ -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 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 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); diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlTaskManager.java b/presto-main/src/main/java/io/prestosql/execution/SqlTaskManager.java index eee8e4df8..283c2de8a 100644 --- a/presto-main/src/main/java/io/prestosql/execution/SqlTaskManager.java +++ b/presto-main/src/main/java/io/prestosql/execution/SqlTaskManager.java @@ -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( diff --git a/presto-main/src/main/java/io/prestosql/split/PageSourceManager.java b/presto-main/src/main/java/io/prestosql/split/PageSourceManager.java index fb80a710c..cf8aae7bd 100644 --- a/presto-main/src/main/java/io/prestosql/split/PageSourceManager.java +++ b/presto-main/src/main/java/io/prestosql/split/PageSourceManager.java @@ -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 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 constraint = null; diff --git a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java index c69776596..8bdedb456 100644 --- a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java +++ b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java @@ -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); diff --git a/presto-main/src/test/java/io/prestosql/catalog/TestDynamicCatalogRunner.java b/presto-main/src/test/java/io/prestosql/catalog/TestDynamicCatalogRunner.java index 3431c1534..f8adfa891 100644 --- a/presto-main/src/test/java/io/prestosql/catalog/TestDynamicCatalogRunner.java +++ b/presto-main/src/test/java/io/prestosql/catalog/TestDynamicCatalogRunner.java @@ -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) diff --git a/presto-main/src/test/java/io/prestosql/execution/EmptyMockMetadata.java b/presto-main/src/test/java/io/prestosql/execution/EmptyMockMetadata.java new file mode 100644 index 000000000..da56a30ea --- /dev/null +++ b/presto-main/src/test/java/io/prestosql/execution/EmptyMockMetadata.java @@ -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 +{ +} diff --git a/presto-main/src/test/java/io/prestosql/execution/TaskTestUtils.java b/presto-main/src/test/java/io/prestosql/execution/TaskTestUtils.java index d8e0773f8..c0457c39f 100644 --- a/presto-main/src/test/java/io/prestosql/execution/TaskTestUtils.java +++ b/presto-main/src/test/java/io/prestosql/execution/TaskTestUtils.java @@ -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 diff --git a/presto-main/src/test/java/io/prestosql/execution/TestMemoryRevokingScheduler.java b/presto-main/src/test/java/io/prestosql/execution/TestMemoryRevokingScheduler.java index 4930c1096..fb98b8f7d 100644 --- a/presto-main/src/test/java/io/prestosql/execution/TestMemoryRevokingScheduler.java +++ b/presto-main/src/test/java/io/prestosql/execution/TestMemoryRevokingScheduler.java @@ -306,6 +306,7 @@ public class TestMemoryRevokingScheduler executor, Functions.identity(), new DataSize(32, MEGABYTE), - new CounterStat()); + new CounterStat(), + new EmptyMockMetadata()); } } diff --git a/presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java b/presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java index e21651bd0..3e0bb5164 100644 --- a/presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java +++ b/presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java @@ -322,6 +322,7 @@ public class TestSqlTask taskNotificationExecutor, Functions.identity(), new DataSize(32, MEGABYTE), - new CounterStat()); + new CounterStat(), + new EmptyMockMetadata()); } } diff --git a/presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java b/presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java index b24735af9..2a520d419 100644 --- a/presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java +++ b/presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java @@ -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 splits, OutputBuffers outputBuffers)