[hotfix][table-common] Fix invalid BYTES data type extraction

spotify · Feb 5, 2020 · 2b2d6ac · 2b2d6ac
1 parent 73017a8
commit 2b2d6ac
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 4 deletions.
diff --git a/...table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java b/...table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
@@ -305,8 +305,12 @@ private DataType extractDataTypeOrError(DataTypeTemplate template, List<Type> ty
  DataTypeTemplate template,
  List<Type> typeHierarchy,
  Type type) {
+ // prefer BYTES over ARRAY<TINYINT> for byte[]
+ if (type == byte[].class) {
+ return DataTypes.BYTES();
+ }
  // for T[]
- if (type instanceof GenericArrayType) {
+ else if (type instanceof GenericArrayType) {
  final GenericArrayType genericArray = (GenericArrayType) type;
  return DataTypes.ARRAY(
  extractDataTypeOrRaw(template, typeHierarchy, genericArray.getGenericComponentType()));

diff --git a/...table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java b/...table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
@@ -19,10 +19,12 @@
 package org.apache.flink.table.types.utils;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.annotation.DataTypeHint;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.expressions.TableSymbol;
 import org.apache.flink.table.types.AtomicDataType;
 import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.extraction.DataTypeExtractor;
 import org.apache.flink.table.types.logical.SymbolType;
 import org.apache.flink.types.Row;
 
@@ -34,6 +36,9 @@
 /**
  * Class-based data type extractor that supports extraction of clearly identifiable data types for
  * input and output conversion.
+ *
+ * <p>Note: In most of the cases, {@link DataTypeExtractor} is more useful as it also considers structured
+ * types and type variables possibly annotated with {@link DataTypeHint}.
  */
 @Internal
 public final class ClassDataTypeConverter {
@@ -87,8 +92,8 @@ private static void addDefaultDataType(Class<?> clazz, DataType rootType) {
  */
  @SuppressWarnings("unchecked")
  public static Optional<DataType> extractDataType(Class<?> clazz) {
- // byte arrays have higher priority than regular arrays
- if (clazz.equals(byte[].class)) {
+ // prefer BYTES over ARRAY<TINYINT> for byte[]
+ if (clazz == byte[].class) {
  return Optional.of(DataTypes.BYTES());
  }
 

diff --git a/...e-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java b/...e-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
@@ -77,11 +77,16 @@ public class DataTypeExtractorTest {
  @Parameters
  public static List<TestSpec> testData() {
  return Arrays.asList(
- // simple extraction
+ // simple extraction of INT
  TestSpec
  .forType(Integer.class)
  .expectDataType(DataTypes.INT()),
 
+ // simple extraction of BYTES
+ TestSpec
+ .forType(byte[].class)
+ .expectDataType(DataTypes.BYTES()),
+
  // extraction from hint conversion class
  TestSpec
  .forType(