[hotfix][connector-hive] Fix Hive type mapping to Table API type information

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));

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
 		};