This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b5cd9f34ab7 [FLINK-29152][hive] Fix inconsistent behavior with Hive
for `desc table` in Hive dialect
b5cd9f34ab7 is described below
commit b5cd9f34ab73fa69a3db5a09908c1aa954ed0597
Author: yuxia Luo <[email protected]>
AuthorDate: Wed Sep 21 23:10:04 2022 +0800
[FLINK-29152][hive] Fix inconsistent behavior with Hive for `desc table` in
Hive dialect
This closes #20789
---
.../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
