This is an automated email from the ASF dual-hosted git repository. jark pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git
commit a6e954ca3bff9c62713d475627b49dd18a4f02fd Author: luoyuxia <[email protected]> AuthorDate: Thu Sep 8 20:01:00 2022 +0800 [FLINK-29152][hive] fix inconsistent behavior with Hive for `desc table` in Hive dialect --- .../e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 | 1 - .../flink/table/catalog/hive/HiveCatalog.java | 2 +- .../delegation/hive/HiveOperationExecutor.java | 82 ++++++++++++++++++++++ .../flink/connectors/hive/HiveDialectITCase.java | 29 ++++++++ .../src/test/resources/endpoint/hive_catalog.q | 16 ++--- 5 files changed, 120 insertions(+), 10 deletions(-) diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 index 4c8647a7c1c..52bf0eb6772 100644 --- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 +++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 @@ -26,4 +26,3 @@ Method <org.apache.flink.streaming.api.operators.SourceOperator$1$1.registerRele Method <org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.getTransactionCoordinatorId()> calls method <org.apache.flink.streaming.connectors.kafka.internals.FlinkKafkaInternalProducer.getTransactionCoordinatorId()> in (FlinkKafkaProducer.java:1327) Method <org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.init()> calls method <org.apache.flink.streaming.api.operators.SourceOperator.getSourceReader()> in (SourceOperatorStreamTask.java:75) Method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.isIdle()> calls method <org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.isDefaultActionAvailable()> in (MailboxExecutorImpl.java:63) -Method <org.apache.flink.table.planner.delegation.hive.parse.HiveParserDDLSemanticAnalyzer.getTable(org.apache.flink.table.catalog.ObjectPath)> calls method <org.apache.flink.table.catalog.hive.HiveCatalog.getHiveTable(org.apache.flink.table.catalog.ObjectPath)> in (HiveParserDDLSemanticAnalyzer.java:273) \ No newline at end of file diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java index b0860af8d3a..b6b3df6ba13 100644 --- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java +++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java @@ -710,7 +710,7 @@ public class HiveCatalog extends AbstractCatalog { } } - @VisibleForTesting + @Internal public Table getHiveTable(ObjectPath tablePath) throws TableNotExistException { try { Table table = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName()); diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java index 07353da1202..0afc190db27 100644 --- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java +++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java @@ -28,10 +28,13 @@ import org.apache.flink.table.api.internal.TableResultInternal; import org.apache.flink.table.catalog.Catalog; import org.apache.flink.table.catalog.CatalogManager; import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.hive.HiveCatalog; import org.apache.flink.table.delegation.ExtendedOperationExecutor; +import org.apache.flink.table.operations.DescribeTableOperation; import org.apache.flink.table.operations.ExplainOperation; import org.apache.flink.table.operations.HiveSetOperation; import org.apache.flink.table.operations.Operation; @@ -39,13 +42,17 @@ import org.apache.flink.table.planner.delegation.PlannerContext; import org.apache.flink.table.planner.delegation.hive.copy.HiveSetProcessor; import org.apache.flink.table.planner.delegation.hive.operations.HiveLoadDataOperation; import org.apache.flink.table.planner.delegation.hive.operations.HiveShowCreateTableOperation; +import org.apache.flink.table.types.DataType; import org.apache.flink.types.Row; +import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.api.FieldSchema; import org.apache.hadoop.hive.ql.metadata.Hive; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.ql.metadata.Table; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; @@ -76,6 +83,8 @@ public class HiveOperationExecutor implements ExtendedOperationExecutor { return executeHiveLoadDataOperation((HiveLoadDataOperation) operation); } else if (operation instanceof HiveShowCreateTableOperation) { return executeShowCreateTableOperation((HiveShowCreateTableOperation) operation); + } else if (operation instanceof DescribeTableOperation) { + return executeDescribeTableOperation((DescribeTableOperation) operation); } else if (operation instanceof ExplainOperation) { ExplainOperation explainOperation = (ExplainOperation) operation; if (explainOperation.getChild() instanceof HiveLoadDataOperation) { @@ -264,4 +273,77 @@ public class HiveOperationExecutor implements ExtendedOperationExecutor { .build(); return Optional.of(resultInternal); } + + private Optional<TableResultInternal> executeDescribeTableOperation( + DescribeTableOperation describeTableOperation) { + // currently, if it's 'describe extended', we still delegate to Flink's own implementation + if (describeTableOperation.isExtended()) { + return Optional.empty(); + } else { + ObjectIdentifier tableIdentifier = describeTableOperation.getSqlIdentifier(); + Catalog currentCatalog = + catalogManager.getCatalog(catalogManager.getCurrentCatalog()).orElse(null); + if (!(currentCatalog instanceof HiveCatalog)) { + // delegate to Flink's own implementation + return Optional.empty(); + } + HiveCatalog hiveCatalog = (HiveCatalog) currentCatalog; + ObjectPath tablePath = + new ObjectPath( + tableIdentifier.getDatabaseName(), tableIdentifier.getObjectName()); + org.apache.hadoop.hive.metastore.api.Table table; + try { + table = hiveCatalog.getHiveTable(tablePath); + } catch (TableNotExistException e) { + throw new FlinkHiveException( + String.format( + "The table or view %s doesn't exist in catalog %s.", + tablePath, catalogManager.getCurrentCatalog()), + e); + } + if (!HiveCatalog.isHiveTable(table.getParameters())) { + // if it's not a Hive table, delegate to Flink's own implementation + return Optional.empty(); + } + List<Row> result = new ArrayList<>(); + // describe table's columns + List<FieldSchema> columns = table.getSd().getCols(); + List<FieldSchema> partitionColumns = table.getPartitionKeys(); + for (FieldSchema fieldSchema : columns) { + result.add(describeColumn(fieldSchema)); + } + for (FieldSchema fieldSchema : partitionColumns) { + result.add(describeColumn(fieldSchema)); + } + + // table's partition information + if (!partitionColumns.isEmpty()) { + result.add(Row.of("# Partition Information", "", "")); + for (FieldSchema fieldSchema : partitionColumns) { + result.add(describeColumn(fieldSchema)); + } + } + TableResultInternal tableResultInternal = + TableResultImpl.builder() + .resultKind(ResultKind.SUCCESS) + .schema( + ResolvedSchema.physical( + new String[] {"col_name", "data_type", "comment"}, + new DataType[] { + DataTypes.STRING(), + DataTypes.STRING(), + DataTypes.STRING() + })) + .data(result) + .build(); + return Optional.of(tableResultInternal); + } + } + + private Row describeColumn(FieldSchema fieldSchema) { + return Row.of( + fieldSchema.getName(), + fieldSchema.getType(), + fieldSchema.getComment() == null ? StringUtils.EMPTY : fieldSchema.getComment()); + } } diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java index 14e32815dbd..a878510f948 100644 --- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java +++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java @@ -1229,6 +1229,35 @@ public class HiveDialectITCase { assertThat(actualResult).isEqualTo(expectedResult); } + @Test + public void testDescribeTable() { + tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT); + tableEnv.executeSql( + "create table t1(id BIGINT,\n" + + " name STRING) WITH (\n" + + " 'connector' = 'datagen' " + + ")"); + tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE); + tableEnv.executeSql("create table t2(a int, b string, c boolean)"); + tableEnv.executeSql( + "create table t3(a decimal(10, 2), b double, c float) partitioned by (d date)"); + + // desc non-hive table + List<Row> result = CollectionUtil.iteratorToList(tableEnv.executeSql("desc t1").collect()); + assertThat(result.toString()) + .isEqualTo( + "[+I[id, BIGINT, true, null, null, null], +I[name, STRING, true, null, null, null]]"); + // desc hive table + result = CollectionUtil.iteratorToList(tableEnv.executeSql("desc t2").collect()); + assertThat(result.toString()) + .isEqualTo("[+I[a, int, ], +I[b, string, ], +I[c, boolean, ]]"); + result = CollectionUtil.iteratorToList(tableEnv.executeSql("desc default.t3").collect()); + assertThat(result.toString()) + .isEqualTo( + "[+I[a, decimal(10,2), ], +I[b, double, ], +I[c, float, ], +I[d, date, ]," + + " +I[# Partition Information, , ], +I[d, date, ]]"); + } + @Test public void testUnsupportedOperation() { List<String> statements = diff --git a/flink-connectors/flink-connector-hive/src/test/resources/endpoint/hive_catalog.q b/flink-connectors/flink-connector-hive/src/test/resources/endpoint/hive_catalog.q index 95f1a36d458..f4aae377010 100644 --- a/flink-connectors/flink-connector-hive/src/test/resources/endpoint/hive_catalog.q +++ b/flink-connectors/flink-connector-hive/src/test/resources/endpoint/hive_catalog.q @@ -102,15 +102,15 @@ show current database; # test hive table with parameterized types # ========================================================================== -describe hive.additional_test_database.param_types_table; +describe additional_test_database.param_types_table; !output -+------+-----------------+------+-----+--------+-----------+ -| name | type | null | key | extras | watermark | -+------+-----------------+------+-----+--------+-----------+ -| dec | DECIMAL(10, 10) | TRUE | | | | -| ch | CHAR(5) | TRUE | | | | -| vch | VARCHAR(15) | TRUE | | | | -+------+-----------------+------+-----+--------+-----------+ ++----------+----------------+---------+ +| col_name | data_type | comment | ++----------+----------------+---------+ +| dec | decimal(10,10) | | +| ch | char(5) | | +| vch | varchar(15) | | ++----------+----------------+---------+ 3 rows in set !ok
