From 9f19345767e930f4a9fcae89313de79688949f90 Mon Sep 17 00:00:00 2001 From: Timo Walther Date: Wed, 22 May 2019 11:12:23 +0200 Subject: [PATCH] [hotfix][connector-hive] Fix Hive type mapping to Table API type information --- .../table/catalog/hive/util/HiveTypeUtil.java | 17 ++++++----------- .../hive/HiveCatalogGenericMetadataTest.java | 6 +++--- 2 files changed, 9 insertions(+), 14 deletions(-) diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java index d26d20bba0f..5a944156244 100644 --- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java +++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java @@ -20,6 +20,7 @@ package org.apache.flink.table.catalog.hive.util; import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo; import org.apache.flink.api.common.typeinfo.TypeInformation; @@ -64,24 +65,18 @@ public class HiveTypeUtil { return serdeConstants.DOUBLE_TYPE_NAME; } else if (type == BasicTypeInfo.STRING_TYPE_INFO) { return serdeConstants.STRING_TYPE_NAME; - } else if (type == BasicTypeInfo.DATE_TYPE_INFO) { + } else if (type == SqlTimeTypeInfo.DATE) { return serdeConstants.DATE_TYPE_NAME; - } else if (type == BasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO) { + } else if (type == PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO) { return serdeConstants.BINARY_TYPE_NAME; - } else if (type instanceof SqlTimeTypeInfo) { + } else if (type == SqlTimeTypeInfo.TIMESTAMP) { return serdeConstants.TIMESTAMP_TYPE_NAME; - } else if (type instanceof BasicArrayTypeInfo) { - return toHiveArrayType((BasicArrayTypeInfo) type); } else { throw new UnsupportedOperationException( String.format("Flink doesn't support converting type %s to Hive type yet.", type.toString())); } } - private static String toHiveArrayType(BasicArrayTypeInfo arrayTypeInfo) { - return String.format(HIVE_ARRAY_TYPE_NAME_FORMAT, toHiveType(arrayTypeInfo.getComponentInfo())); - } - /** * Convert Hive data type to a Flink data type. * TODO: the following Hive types are not supported in Flink yet, including CHAR, VARCHAR, DECIMAL, MAP, STRUCT @@ -127,11 +122,11 @@ public class HiveTypeUtil { case DOUBLE: return BasicTypeInfo.DOUBLE_TYPE_INFO; case DATE: - return BasicTypeInfo.DATE_TYPE_INFO; + return SqlTimeTypeInfo.DATE; case TIMESTAMP: return SqlTimeTypeInfo.TIMESTAMP; case BINARY: - return BasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO; + return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO; default: throw new UnsupportedOperationException( String.format("Flink doesn't support Hive primitive type %s yet", hiveType)); diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogGenericMetadataTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogGenericMetadataTest.java index ae5f5c89703..9a35068acb2 100644 --- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogGenericMetadataTest.java +++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogGenericMetadataTest.java @@ -18,8 +18,8 @@ package org.apache.flink.table.catalog.hive; -import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.table.api.TableSchema; @@ -65,8 +65,8 @@ public class HiveCatalogGenericMetadataTest extends CatalogTestBase { BasicTypeInfo.DOUBLE_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, - BasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO, - BasicTypeInfo.DATE_TYPE_INFO, + PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO, + SqlTimeTypeInfo.DATE, SqlTimeTypeInfo.TIMESTAMP }; -- GitLab