You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2022/08/04 12:35:52 UTC
[flink] 02/02: [FLINK-28629][sql-gateway][hive] Allow to getCatalogs in the HiveServer2 Endpoint
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e3b26c283ca3ad266978b719d21e2da72ff225da
Author: Shengkai <10...@qq.com>
AuthorDate: Tue Aug 2 12:18:19 2022 +0800
[FLINK-28629][sql-gateway][hive] Allow to getCatalogs in the HiveServer2 Endpoint
This closes #20334
---
.../table/endpoint/hive/HiveServer2Endpoint.java | 29 +++++++++++++++-
.../table/endpoint/hive/HiveServer2Schemas.java | 25 +++++++++-----
.../endpoint/hive/util/StringRowDataUtils.java | 24 ++++++++-----
.../hive/util/ThriftObjectConversions.java | 2 ++
.../endpoint/hive/HiveServer2EndpointITCase.java | 40 ++++++++++++++++++++++
.../flink/table/gateway/api/SqlGatewayService.java | 15 +++++++-
.../table/gateway/api/operation/OperationType.java | 3 ++
.../gateway/api/utils/MockedSqlGatewayService.java | 6 ++++
.../gateway/service/SqlGatewayServiceImpl.java | 11 ++++++
.../service/operation/OperationExecutor.java | 7 ++++
.../table/gateway/service/session/Session.java | 4 +++
.../gateway/service/SqlGatewayServiceITCase.java | 12 +++++++
12 files changed, 158 insertions(+), 20 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Endpoint.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Endpoint.java
index 8ae5bb23954..0772a4e0b24 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Endpoint.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Endpoint.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
+import org.apache.flink.table.endpoint.hive.util.StringRowDataUtils;
import org.apache.flink.table.gateway.api.SqlGatewayService;
import org.apache.flink.table.gateway.api.endpoint.EndpointVersion;
import org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpoint;
@@ -112,6 +113,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -119,6 +121,7 @@ import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYNC;
import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_SQL_DIALECT;
import static org.apache.flink.table.endpoint.hive.HiveServer2EndpointVersion.HIVE_CLI_SERVICE_PROTOCOL_V10;
+import static org.apache.flink.table.endpoint.hive.HiveServer2Schemas.GET_CATALOGS_SCHEMA;
import static org.apache.flink.table.endpoint.hive.util.HiveJdbcParameterUtils.getUsedDefaultDatabase;
import static org.apache.flink.table.endpoint.hive.util.HiveJdbcParameterUtils.validateAndNormalize;
import static org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions.toFetchOrientation;
@@ -130,6 +133,7 @@ import static org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions.
import static org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions.toTSessionHandle;
import static org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions.toTStatus;
import static org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions.toTTableSchema;
+import static org.apache.flink.table.gateway.api.results.ResultSet.ResultType.EOS;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -390,7 +394,30 @@ public class HiveServer2Endpoint implements TCLIService.Iface, SqlGatewayEndpoin
@Override
public TGetCatalogsResp GetCatalogs(TGetCatalogsReq tGetCatalogsReq) throws TException {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
+ TGetCatalogsResp resp = new TGetCatalogsResp();
+ try {
+ SessionHandle sessionHandle = toSessionHandle(tGetCatalogsReq.getSessionHandle());
+ OperationHandle operationHandle =
+ service.submitOperation(
+ sessionHandle,
+ OperationType.LIST_CATALOGS,
+ () -> {
+ Set<String> catalogNames = service.listCatalogs(sessionHandle);
+ return new ResultSet(
+ EOS,
+ null,
+ GET_CATALOGS_SCHEMA,
+ StringRowDataUtils.toRowData(catalogNames));
+ });
+ resp.setStatus(OK_STATUS);
+ resp.setOperationHandle(
+ toTOperationHandle(
+ sessionHandle, operationHandle, OperationType.LIST_CATALOGS));
+ } catch (Throwable t) {
+ LOG.error("Failed to GetCatalogs.", t);
+ resp.setStatus(toTStatus(t));
+ }
+ return resp;
}
@Override
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Schemas.java
similarity index 51%
copy from flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
copy to flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Schemas.java
index babe99fd3f9..0a6818330fd 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/HiveServer2Schemas.java
@@ -16,16 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.table.gateway.api.operation;
+package org.apache.flink.table.endpoint.hive;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
-/** The Operation Type. */
-@PublicEvolving
-public enum OperationType {
- /** The type indicates the operation executes statements. */
- EXECUTE_STATEMENT,
+import java.util.Collections;
- /** The type indicates the operation is unknown. */
- UNKNOWN;
+/** Schemas for the HiveServer2 Endpoint result. */
+public class HiveServer2Schemas {
+
+ /** Schema for {@link HiveServer2Endpoint#GetCatalogs}. */
+ public static final ResolvedSchema GET_CATALOGS_SCHEMA =
+ new ResolvedSchema(
+ Collections.singletonList(
+ Column.physical("TABLE_CAT", DataTypes.STRING())
+ .withComment("Catalog name. NULL if not applicable.")),
+ Collections.emptyList(),
+ null);
}
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/StringRowDataUtils.java
similarity index 57%
copy from flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
copy to flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/StringRowDataUtils.java
index babe99fd3f9..103f7311c89 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/StringRowDataUtils.java
@@ -16,16 +16,22 @@
* limitations under the License.
*/
-package org.apache.flink.table.gateway.api.operation;
+package org.apache.flink.table.endpoint.hive.util;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
-/** The Operation Type. */
-@PublicEvolving
-public enum OperationType {
- /** The type indicates the operation executes statements. */
- EXECUTE_STATEMENT,
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
- /** The type indicates the operation is unknown. */
- UNKNOWN;
+/** Convert the {@link String} to {@link RowData}. */
+public class StringRowDataUtils {
+
+ public static List<RowData> toRowData(Collection<String> external) {
+ return external.stream()
+ .map(val -> GenericRowData.of(StringData.fromString(val)))
+ .collect(Collectors.toList());
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/ThriftObjectConversions.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/ThriftObjectConversions.java
index eccf3c7e1f4..235796e4eb3 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/ThriftObjectConversions.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/endpoint/hive/util/ThriftObjectConversions.java
@@ -160,6 +160,8 @@ public class ThriftObjectConversions {
switch (type) {
case EXECUTE_STATEMENT:
return TOperationType.EXECUTE_STATEMENT;
+ case LIST_CATALOGS:
+ return TOperationType.GET_CATALOGS;
case UNKNOWN:
return TOperationType.UNKNOWN;
default:
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/endpoint/hive/HiveServer2EndpointITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/endpoint/hive/HiveServer2EndpointITCase.java
index 269e1a21f3b..7b567e184c0 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/endpoint/hive/HiveServer2EndpointITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/endpoint/hive/HiveServer2EndpointITCase.java
@@ -19,8 +19,12 @@
package org.apache.flink.table.endpoint.hive;
import org.apache.flink.core.testutils.FlinkAssertions;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.hive.util.HiveTypeUtil;
import org.apache.flink.table.endpoint.hive.util.HiveServer2EndpointExtension;
import org.apache.flink.table.endpoint.hive.util.ThriftObjectConversions;
import org.apache.flink.table.gateway.api.operation.OperationHandle;
@@ -38,6 +42,7 @@ import org.apache.flink.util.function.BiConsumerWithException;
import org.apache.flink.util.function.ThrowingConsumer;
import org.apache.hadoop.hive.common.auth.HiveAuthUtils;
+import org.apache.hive.jdbc.JdbcColumn;
import org.apache.hive.service.rpc.thrift.TCLIService;
import org.apache.hive.service.rpc.thrift.TCancelOperationReq;
import org.apache.hive.service.rpc.thrift.TCancelOperationResp;
@@ -57,8 +62,11 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import java.net.InetAddress;
import java.sql.Connection;
+import java.sql.ResultSetMetaData;
import java.util.AbstractMap;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -186,6 +194,23 @@ public class HiveServer2EndpointITCase extends TestLogger {
operationHandle)))));
}
+ @Test
+ public void testGetCatalogs() throws Exception {
+ try (Connection connection = ENDPOINT_EXTENSION.getConnection()) {
+ java.sql.ResultSet result = connection.getMetaData().getCatalogs();
+ assertSchemaEquals(
+ ResolvedSchema.of(Column.physical("TABLE_CAT", DataTypes.STRING())),
+ result.getMetaData());
+
+ List<String> actual = new ArrayList<>();
+ while (result.next()) {
+ actual.add(result.getString(1));
+ }
+
+ assertThat(actual).contains("hive", "default_catalog");
+ }
+ }
+
private void runOperationRequest(
ThrowingConsumer<TOperationHandle, Exception> manipulateOp,
BiConsumerWithException<SessionHandle, OperationHandle, Exception> operationValidator)
@@ -225,4 +250,19 @@ public class HiveServer2EndpointITCase extends TestLogger {
transport.open();
return new TCLIService.Client(new TBinaryProtocol(transport));
}
+
+ private void assertSchemaEquals(ResolvedSchema expected, ResultSetMetaData metaData)
+ throws Exception {
+ assertThat(metaData.getColumnCount()).isEqualTo(expected.getColumnCount());
+ for (int i = 1; i <= metaData.getColumnCount(); i++) {
+ Column column =
+ expected.getColumn(i - 1)
+ .orElseThrow(() -> new RuntimeException("Can not get column."));
+ assertThat(metaData.getColumnName(i)).isEqualTo(column.getName());
+ int jdbcType =
+ JdbcColumn.hiveTypeToSqlType(
+ HiveTypeUtil.toHiveTypeInfo(column.getDataType(), false).getTypeName());
+ assertThat(metaData.getColumnType(i)).isEqualTo(jdbcType);
+ }
+ }
}
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java
index fd2ed677243..97b377c59c4 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java
+++ b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java
@@ -33,6 +33,7 @@ import org.apache.flink.table.gateway.api.session.SessionHandle;
import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
/** A service of SQL gateway is responsible for handling requests from the endpoints. */
@@ -135,7 +136,7 @@ public interface SqlGatewayService {
throws SqlGatewayException;
// -------------------------------------------------------------------------------------------
- // Statements
+ // Statements API
// -------------------------------------------------------------------------------------------
/**
@@ -185,4 +186,16 @@ public interface SqlGatewayService {
OperationHandle operationHandle,
FetchOrientation orientation,
int maxRows);
+
+ // -------------------------------------------------------------------------------------------
+ // Catalog API
+ // -------------------------------------------------------------------------------------------
+
+ /**
+ * Return all available catalogs in the current session.
+ *
+ * @param sessionHandle handle to identify the session.
+ * @return names of the registered catalogs.
+ */
+ Set<String> listCatalogs(SessionHandle sessionHandle) throws SqlGatewayException;
}
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
index babe99fd3f9..6177008b264 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
+++ b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationType.java
@@ -26,6 +26,9 @@ public enum OperationType {
/** The type indicates the operation executes statements. */
EXECUTE_STATEMENT,
+ /** The type indicates the operation list catalogs. */
+ LIST_CATALOGS,
+
/** The type indicates the operation is unknown. */
UNKNOWN;
}
diff --git a/flink-table/flink-sql-gateway-api/src/test/java/org/apache/flink/table/gateway/api/utils/MockedSqlGatewayService.java b/flink-table/flink-sql-gateway-api/src/test/java/org/apache/flink/table/gateway/api/utils/MockedSqlGatewayService.java
index a75acd72566..af0ff36ec6e 100644
--- a/flink-table/flink-sql-gateway-api/src/test/java/org/apache/flink/table/gateway/api/utils/MockedSqlGatewayService.java
+++ b/flink-table/flink-sql-gateway-api/src/test/java/org/apache/flink/table/gateway/api/utils/MockedSqlGatewayService.java
@@ -31,6 +31,7 @@ import org.apache.flink.table.gateway.api.session.SessionEnvironment;
import org.apache.flink.table.gateway.api.session.SessionHandle;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
/** Mocked {@link SqlGatewayService}. */
@@ -114,4 +115,9 @@ public class MockedSqlGatewayService implements SqlGatewayService {
throws SqlGatewayException {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public Set<String> listCatalogs(SessionHandle sessionHandle) throws SqlGatewayException {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java
index 3f6678e45e4..9419b6071c3 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java
@@ -38,6 +38,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
/** The implementation of the {@link SqlGatewayService} interface. */
@@ -210,6 +211,16 @@ public class SqlGatewayServiceImpl implements SqlGatewayService {
}
}
+ @Override
+ public Set<String> listCatalogs(SessionHandle sessionHandle) throws SqlGatewayException {
+ try {
+ return getSession(sessionHandle).createExecutor().listCatalogs();
+ } catch (Throwable t) {
+ LOG.error("Failed to listCatalogs.", t);
+ throw new SqlGatewayException("Failed to listCatalogs.", t);
+ }
+ }
+
@VisibleForTesting
Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
index b54afea608e..e49a91f2ae8 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
@@ -46,6 +46,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.flink.table.gateway.service.utils.Constants.JOB_ID;
import static org.apache.flink.table.gateway.service.utils.Constants.SET_KEY;
@@ -99,6 +100,12 @@ public class OperationExecutor {
}
}
+ public Set<String> listCatalogs() {
+ return getTableEnvironment().getCatalogManager().listCatalogs();
+ }
+
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public TableEnvironmentInternal getTableEnvironment() {
TableEnvironmentInternal tableEnv = sessionContext.createTableEnvironment();
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
index 74689d7e5ac..87d77a87c36 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
@@ -66,6 +66,10 @@ public class Session implements Closeable {
return sessionContext.getOperationManager();
}
+ public OperationExecutor createExecutor() {
+ return sessionContext.createOperationExecutor(new Configuration());
+ }
+
public OperationExecutor createExecutor(Configuration executionConfig) {
return sessionContext.createOperationExecutor(executionConfig);
}
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java
index 9cf33996b04..69880a74dec 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java
@@ -304,6 +304,18 @@ public class SqlGatewayServiceITCase extends AbstractTestBase {
task -> assertThat(task.get()).isEqualTo(getDefaultResultSet().getResultSchema()));
}
+ @Test
+ public void testListCatalogs() {
+ SessionEnvironment environment =
+ SessionEnvironment.newBuilder()
+ .setSessionEndpointVersion(MockedEndpointVersion.V1)
+ .registerCatalog("cat1", new GenericInMemoryCatalog("cat1"))
+ .registerCatalog("cat2", new GenericInMemoryCatalog("cat2"))
+ .build();
+ SessionHandle sessionHandle = service.openSession(environment);
+ assertThat(service.listCatalogs(sessionHandle)).contains("cat1", "cat2");
+ }
+
// --------------------------------------------------------------------------------------------
// Concurrent tests
// --------------------------------------------------------------------------------------------