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 <[email protected]> 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 // --------------------------------------------------------------------------------------------
