You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2019/12/02 15:54:18 UTC
[flink] branch master updated: [FLINK-14967][table] Add a utility
for creating data types via reflection
This is an automated email from the ASF dual-hosted git repository.
twalthr 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 a43ae47 [FLINK-14967][table] Add a utility for creating data types via reflection
a43ae47 is described below
commit a43ae47a0c2998d35ec1f9970cd091bd90952e04
Author: Timo Walther <tw...@apache.org>
AuthorDate: Wed Nov 27 18:44:21 2019 +0100
[FLINK-14967][table] Add a utility for creating data types via reflection
This implements the data type extractor mentioned in FLIP-65. It is similar to
Flink's core type information extractor but also adds a lot of SQL specific features
and improves the overall user experience. In particular, it allows to annotate
types, fields, and classes for parameterizing the extraction process. It is unified
across Java and Scala.
This closes #10342.
---
flink-table/flink-table-api-scala/pom.xml | 12 +
.../extraction/DataTypeExtractorScalaTest.scala | 91 +++
flink-table/flink-table-common/pom.xml | 7 +
.../flink/table/annotation/DataTypeHint.java | 268 +++++++
.../flink/table/annotation/ExtractionVersion.java | 42 ++
.../apache/flink/table/annotation/HintFlag.java | 34 +
.../apache/flink/table/annotation/InputGroup.java | 47 ++
.../flink/table/annotation/UnknownSerializer.java | 33 +
.../apache/flink/table/catalog/DataTypeLookup.java | 54 ++
.../table/types/extraction/DataTypeExtractor.java | 538 ++++++++++++++
.../types/extraction/utils/DataTypeTemplate.java | 384 ++++++++++
.../types/extraction/utils/ExtractionUtils.java | 571 +++++++++++++++
.../table/types/utils/ClassDataTypeConverter.java | 3 +-
.../table/types/ClassDataTypeConverterTest.java | 12 +
.../types/extraction/DataTypeExtractorTest.java | 815 +++++++++++++++++++++
.../types/extraction/utils/DataTypeHintMock.java | 112 +++
.../types/inference/utils/DataTypeLookupMock.java | 55 ++
17 files changed, 3077 insertions(+), 1 deletion(-)
diff --git a/flink-table/flink-table-api-scala/pom.xml b/flink-table/flink-table-api-scala/pom.xml
index 2132f95..be81642 100644
--- a/flink-table/flink-table-api-scala/pom.xml
+++ b/flink-table/flink-table-api-scala/pom.xml
@@ -48,7 +48,9 @@ under the License.
<artifactId>flink-table-api-java</artifactId>
<version>${project.version}</version>
</dependency>
+
<!-- External dependencies -->
+
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-reflect</artifactId>
@@ -61,6 +63,16 @@ under the License.
<groupId>org.scala-lang</groupId>
<artifactId>scala-compiler</artifactId>
</dependency>
+
+ <!-- Test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/types/extraction/DataTypeExtractorScalaTest.scala b/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/types/extraction/DataTypeExtractorScalaTest.scala
new file mode 100644
index 0000000..af739cf
--- /dev/null
+++ b/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/types/extraction/DataTypeExtractorScalaTest.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction
+
+import java.util
+
+import org.apache.flink.table.annotation.{DataTypeHint, HintFlag}
+import org.apache.flink.table.types.extraction.DataTypeExtractorScalaTest.{ScalaComplexPojo, ScalaPojoWithCustomFieldOrder, ScalaSimplePojo, ScalaSimplePojoWithManyAnnotations}
+import org.apache.flink.table.types.extraction.DataTypeExtractorTest._
+import org.junit.Test
+
+/**
+ * Scala tests for [[DataTypeExtractor]].
+ */
+class DataTypeExtractorScalaTest {
+
+ val parameters: Seq[TestSpec] = Seq(
+ // simple structured type without RAW type
+ TestSpec
+ .forType(classOf[ScalaSimplePojo])
+ .expectDataType(getSimplePojoDataType(classOf[ScalaSimplePojo])),
+
+ // complex nested structured type annotation on top of type
+ TestSpec
+ .forType(classOf[ScalaComplexPojo])
+ .lookupExpects(classOf[Any])
+ .expectDataType(getComplexPojoDataType(classOf[ScalaComplexPojo], classOf[ScalaSimplePojo])),
+
+ // assigning constructor defines field order
+ TestSpec
+ .forType(classOf[ScalaPojoWithCustomFieldOrder])
+ .expectDataType(getPojoWithCustomOrderDataType(classOf[ScalaPojoWithCustomFieldOrder])),
+
+ // many annotations that partially override each other
+ TestSpec
+ .forType(classOf[ScalaSimplePojoWithManyAnnotations])
+ .expectDataType(getSimplePojoDataType(classOf[ScalaSimplePojoWithManyAnnotations]))
+ )
+
+ @Test
+ def testScalaExtraction(): Unit = {
+ parameters.foreach(runExtraction)
+ }
+}
+
+object DataTypeExtractorScalaTest {
+
+ case class ScalaSimplePojo(
+ intField: Integer,
+ primitiveBooleanField: Boolean,
+ primitiveIntField: Int,
+ stringField: String
+ )
+
+ @DataTypeHint(allowRawGlobally = HintFlag.TRUE)
+ case class ScalaComplexPojo(
+ var mapField: util.Map[String, Integer],
+ var simplePojoField: ScalaSimplePojo,
+ var someObject: Any
+ )
+
+ case class ScalaPojoWithCustomFieldOrder(
+ z: java.lang.Long,
+ y: java.lang.Boolean,
+ x: java.lang.Integer
+ )
+
+ @DataTypeHint(forceRawPattern = Array("java.lang."))
+ class ScalaSimplePojoWithManyAnnotations {
+ @DataTypeHint("INT") var intField: Integer = _
+ var primitiveBooleanField: Boolean = _
+ @DataTypeHint(value = "INT NOT NULL", bridgedTo = classOf[Int]) var primitiveIntField: Any = _
+ @DataTypeHint(forceRawPattern = Array()) var stringField: String = _
+ }
+}
diff --git a/flink-table/flink-table-common/pom.xml b/flink-table/flink-table-common/pom.xml
index c0f112b..19f0f35 100644
--- a/flink-table/flink-table-common/pom.xml
+++ b/flink-table/flink-table-common/pom.xml
@@ -44,6 +44,13 @@ under the License.
<version>${project.version}</version>
</dependency>
+ <!-- Used for structured types extraction. -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-shaded-asm-7</artifactId>
+ <!-- managed version -->
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
new file mode 100644
index 0000000..cf6fe9e
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.annotation;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * A hint that influences the reflection-based extraction of a {@link DataType}.
+ *
+ * <p>Data type hints can parameterize or replace the default extraction logic of individual function parameters
+ * and return types, structured classes, or fields of structured classes. An implementer can choose to
+ * what extent the default extraction logic should be modified.
+ *
+ * <p>The following examples show how to explicitly specify data types, how to parameterize the extraction
+ * logic, or how to accept any data type as an input data type:
+ *
+ * <p>{@code @DataTypeHint("INT")} defines an INT data type with a default conversion class.
+ *
+ * <p>{@code @DataTypeHint(value = "TIMESTAMP(3)", bridgedTo = java.sql.Timestamp.class)} defines a TIMESTAMP
+ * data type of millisecond precision with an explicit conversion class.
+ *
+ * <p>{@code @DataTypeHint(value = "RAW", rawSerializer = MyCustomSerializer.class)} defines a RAW data type
+ * with a custom serializer class.
+ *
+ * <p>{@code @DataTypeHint(version = V1, allowRawGlobally = TRUE)} parameterizes the extraction by requesting
+ * a extraction logic version of 1 and allowing the RAW data type in this structured type (and possibly
+ * nested fields).
+ *
+ * <p>{@code @DataTypeHint(bridgedTo = MyPojo.class, allowRawGlobally = TRUE)} defines that a type should be
+ * extracted from the given conversion class but with parameterized extraction for allowing RAW types.
+ *
+ * <p>{@code @DataTypeHint(inputGroup = ANY)} defines that the input validation should accept any
+ * data type.
+ *
+ * <p>Note: All hint parameters are optional. Hint parameters defined on top of a structured type are
+ * inherited by all (deeply) nested fields unless annotated differently. For example, all occurrences of
+ * {@link java.math.BigDecimal} will be extracted as {@code DECIMAL(12, 2)} if the enclosing structured
+ * class is annotated with {@code @DataTypeHint(defaultDecimalPrecision = 12, defaultDecimalScale = 2)}. Individual
+ * field annotations allow to deviate from those default values.
+ */
+@PublicEvolving
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE, ElementType.METHOD, ElementType.FIELD, ElementType.PARAMETER})
+public @interface DataTypeHint {
+
+ // Note to implementers:
+ // Because "null" is not supported as an annotation value. Every annotation parameter *must* have
+ // some representation for unknown values in order to merge multi-level annotations.
+
+ // --------------------------------------------------------------------------------------------
+ // Explicit data type specification
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * The explicit string representation of a data type. See {@link DataTypes} for a list of supported
+ * data types. For example, {@code INT} for an integer data type or {@code DECIMAL(12, 5)} for decimal
+ * data type with precision 12 and scale 5.
+ *
+ * <p>Use an unparameterized {@code RAW} string for explicitly declaring an opaque data type. For
+ * Flink's default RAW serializer, use {@code @DataTypeHint("RAW")}. For a custom RAW serializer,
+ * use {@code @DataTypeHint(value = "RAW", rawSerializer = MyCustomSerializer.class)}.
+ *
+ * <p>By default, the empty string represents an undefined data type which means that it will be
+ * derived automatically.
+ *
+ * <p>Use {@link #inputGroup()} for accepting a group of similar data types if this hint is used
+ * to enrich input arguments.
+ *
+ * @see LogicalType#asSerializableString()
+ * @see DataTypes
+ */
+ String value() default "";
+
+ /**
+ * Adds a hint that data should be represented using the given class when entering or leaving
+ * the table ecosystem.
+ *
+ * <p>If an explicit data type has been defined via {@link #value()}, a supported conversion class
+ * depends on the logical type and its nullability property.
+ *
+ * <p>If an explicit data type has not been defined via {@link #value()}, this class is used for
+ * reflective extraction of a data type.
+ *
+ * <p>Please see the implementation of {@link LogicalType#supportsInputConversion(Class)},
+ * {@link LogicalType#supportsOutputConversion(Class)}, or the documentation for more information
+ * about supported conversions.
+ *
+ * <p>By default, the conversion class is reflectively extracted.
+ *
+ * @see DataType#bridgedTo(Class)
+ */
+ Class<?> bridgedTo() default void.class;
+
+ /**
+ * Adds a hint that defines a custom serializer that should be used for serializing and deserializing
+ * opaque RAW types. It is used if {@link #value()} is explicitly defined as an unparameterized {@code RAW}
+ * string or if (possibly nested) fields in a structured type need to be handled as an opaque type.
+ *
+ * <p>By default, Flink's default RAW serializer is used.
+ *
+ * @see DataTypes#RAW(Class, TypeSerializer)
+ */
+ Class<? extends TypeSerializer<?>> rawSerializer() default UnknownSerializer.class;
+
+ // --------------------------------------------------------------------------------------------
+ // Group of data types specification
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * This hint parameter influences the extraction of a {@link TypeInference} in functions. It adds
+ * a hint for accepting pre-defined groups of similar types, i.e., more than just one explicit data type.
+ *
+ * <p>Note: This hint parameter is only interpreted when used in function hints or next to arguments
+ * of implementation methods. It has highest precedence above all other hint parameter.
+ *
+ * <p>Some examples:
+ * <pre>
+ * {@code
+ * // expects an integer for the first input argument and allows any data type for the second
+ * @FunctionHint(
+ * input = [@DataTypeHint("INT"), @DataTypeHint(inputGroup = ANY)],
+ * output = @DataTypeHint("BOOLEAN")
+ * )
+ *
+ * // expects an integer for the first input argument and allows any data type for the second
+ * eval(int i, @DataTypeHint(inputGroup = ANY) Object o)
+ * }
+ * </pre>
+ */
+ InputGroup inputGroup() default InputGroup.UNKNOWN;
+
+ // --------------------------------------------------------------------------------------------
+ // Parameterization of the reflection-based extraction
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Version that describes the expected behavior of the reflection-based data type extraction.
+ *
+ * <p>It is meant for future backward compatibility. Whenever the extraction logic is changed,
+ * old function and structured type classes should still return the same data type as before when
+ * versioned accordingly.
+ *
+ * <p>By default, the version is always the most recent one.
+ */
+ ExtractionVersion version() default ExtractionVersion.UNKNOWN;
+
+ /**
+ * Defines that a RAW data type may be used for all classes that cannot be mapped to any SQL-like
+ * data type or cause an error.
+ *
+ * <p>By default, this parameter is set to {@code false} which means that an exception is thrown
+ * for unmapped types. This is helpful to identify and fix faulty implementations. It is generally
+ * recommended to use SQL-like types instead of enabling RAW opaque types.
+ *
+ * <p>If RAW types cannot be avoided, they should be enabled only in designated areas (i.e., within
+ * package prefixes using {@link #allowRawPattern()}) in order to not swallow all errors. However,
+ * this parameter globally enables RAW types for the annotated class and all nested fields.
+ *
+ * <p>This parameter has higher precedence than {@link #allowRawPattern()}.
+ *
+ * @see DataTypes#RAW(Class, TypeSerializer)
+ */
+ HintFlag allowRawGlobally() default HintFlag.UNKNOWN;
+
+ /**
+ * Defines that a RAW data type may be used for all classes that cannot be mapped to any SQL-like
+ * data type or cause an error if their class name starts with or is equal to one of the given patterns.
+ *
+ * <p>For example, if some Joda time classes cannot be mapped to any SQL-like data type, one can
+ * define the class prefix {@code "org.joda.time"}. Some classes might be handled as structured types
+ * on a best effort basis but others will be RAW data types if necessary.
+ *
+ * <p>By default, the pattern list is empty which means that an exception is thrown for unmapped
+ * types. This is helpful to identify and fix faulty implementations. It is generally recommended
+ * to use SQL-like types instead of enabling RAW opaque types.
+ *
+ * <p>If RAW types cannot be avoided, this parameter should be used to enabled them only in designated
+ * areas (i.e., within package prefixes) in order to not swallow all errors.
+ *
+ * <p>This parameter has lower precedence than {@link #allowRawGlobally()} which would globally allow
+ * RAW types in the annotated class and all nested fields.
+ *
+ * @see DataTypes#RAW(Class, TypeSerializer)
+ */
+ String[] allowRawPattern() default {""};
+
+ /**
+ * Defines that a RAW data type must be used for all classes if their class name starts with or
+ * is equal to one of the given patterns.
+ *
+ * <p>For example, one can define the class prefix {@code "org.joda.time", "java.math.BigDecimal"} which
+ * means that all Joda time classes and Java's {@link java.math.BigDecimal} will be handled as RAW data
+ * types regardless if they could be mapped to a more SQL-like data type.
+ *
+ * <p>By default, the pattern list is empty which means that an exception is thrown for unmapped
+ * types. This is helpful to identify and fix faulty implementations. It is generally recommended
+ * to use SQL-like types instead of enabling RAW opaque types.
+ *
+ * <p>If RAW types cannot be avoided, they should be enabled only in designated areas (i.e., within
+ * package prefixes) in order to not swallow all errors. However, compared to {@link #allowRawPattern()},
+ * this parameter forces to skip the extraction entirely for the given prefixes instead of trying
+ * to match a class to a more SQL-like data type.
+ *
+ * <p>This parameter has the highest precedence of all data type related hint parameters.
+ *
+ * @see DataTypes#RAW(Class, TypeSerializer)
+ */
+ String[] forceRawPattern() default {""};
+
+ /**
+ * Defines a default precision for all decimal data types that are extracted.
+ *
+ * <p>By default, decimals are not extracted from classes such as {@link java.math.BigDecimal} because
+ * they don't define a fixed precision and scale which is required in the SQL type system.
+ */
+ int defaultDecimalPrecision() default -1;
+
+ /**
+ * Defines a default scale for all decimal data types that are extracted.
+ *
+ * <p>By default, decimals are not extracted from classes such as {@link java.math.BigDecimal} because
+ * they don't define a fixed precision and scale which is required in the SQL type system.
+ */
+ int defaultDecimalScale() default -1;
+
+ /**
+ * Defines a default year precision for all year-month intervals that are extracted. If set to {@code 0},
+ * an {@code INTERVAL MONTH} data type is extracted.
+ *
+ * <p>By default, {@code INTERVAL YEAR(4) TO MONTH} data types are extracted from classes such as
+ * {@link java.time.Period}.
+ */
+ int defaultYearPrecision() default -1;
+
+ /**
+ * Defines a default fractional second precision for all day-time intervals and timestamps that
+ * are extracted.
+ *
+ * <p>By default, those data types are extracted with nano second precision.
+ */
+ int defaultSecondPrecision() default -1;
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ExtractionVersion.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ExtractionVersion.java
new file mode 100644
index 0000000..adb3356
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ExtractionVersion.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.annotation;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Logical version that describes the expected behavior of the reflection-based data type extraction.
+ *
+ * <p>This enumeration is meant for future backward compatibility. Whenever the extraction logic is changed,
+ * old function and structured type classes should still return the same data type as before when versioned
+ * accordingly.
+ */
+@PublicEvolving
+public enum ExtractionVersion {
+
+ /**
+ * Default if no version is specified.
+ */
+ UNKNOWN,
+
+ /**
+ * Initial reflection-based extraction logic according to FLIP-65.
+ */
+ V1
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/HintFlag.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/HintFlag.java
new file mode 100644
index 0000000..c0f364f
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/HintFlag.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.annotation;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Three-valued flag for representing {@code TRUE}, {@code FALSE}, and {@code UNKNOWN}.
+ */
+@PublicEvolving
+public enum HintFlag {
+
+ TRUE,
+
+ FALSE,
+
+ UNKNOWN
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java
new file mode 100644
index 0000000..1a9034a
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.annotation;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.types.inference.InputTypeValidators;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+/**
+ * A list of commonly used pre-defined groups of similar types for accepting more than just one data
+ * type as an input argument in {@link DataTypeHint}s.
+ *
+ * <p>This list exposes a combination of {@link LogicalTypeFamily} and {@link InputTypeValidators}
+ * via annotations for convenient inline usage.
+ */
+@PublicEvolving
+public enum InputGroup {
+
+ /**
+ * Default if no group is specified.
+ */
+ UNKNOWN,
+
+ /**
+ * Enables input wildcards. Any data type can be passed. The behavior is equal to {@link InputTypeValidators#ANY}.
+ *
+ * <p>Note: The class of the annotated element must be {@link Object} as this is the super class
+ * of all possibly passed data types.
+ */
+ ANY
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/UnknownSerializer.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/UnknownSerializer.java
new file mode 100644
index 0000000..ad38230
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/UnknownSerializer.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.annotation;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/**
+ * Helper class for {@link DataTypeHint} for representing an unknown serializer that should be
+ * replaced with a more specific class.
+ */
+@Internal
+abstract class UnknownSerializer extends TypeSerializer<Object> {
+
+ private UnknownSerializer() {
+ // no instantiation
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DataTypeLookup.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DataTypeLookup.java
new file mode 100644
index 0000000..7244374
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DataTypeLookup.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.RawType;
+import org.apache.flink.table.types.logical.StructuredType;
+
+import java.util.Optional;
+
+/**
+ * Catalog of types that can resolve the name of type to a {@link DataType}. This includes both
+ * built-in logical types as well as user-defined logical types such as {@link DistinctType} and
+ * {@link StructuredType}.
+ */
+@PublicEvolving
+public interface DataTypeLookup {
+
+ /**
+ * Lookup a type by a fully or partially defined name.
+ */
+ Optional<DataType> lookupDataType(String name);
+
+ /**
+ * Lookup a type by an unresolved identifier.
+ */
+ Optional<DataType> lookupDataType(UnresolvedIdentifier identifier);
+
+ /**
+ * Resolves a RAW type for the given class.
+ *
+ * <p>The {@link RawType} requires an instantiated serializer. Flink's default RAW serializer is
+ * configured during the resolution process.
+ */
+ DataType resolveRawDataType(Class<?> clazz);
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
new file mode 100644
index 0000000..c5d111b
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
@@ -0,0 +1,538 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction;
+
+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.catalog.DataTypeLookup;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.extraction.utils.DataTypeTemplate;
+import org.apache.flink.table.types.extraction.utils.ExtractionUtils;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RawType;
+import org.apache.flink.table.types.logical.StructuredType;
+import org.apache.flink.table.types.logical.StructuredType.StructuredAttribute;
+import org.apache.flink.table.types.utils.ClassDataTypeConverter;
+
+import javax.annotation.Nullable;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.GenericArrayType;
+import java.lang.reflect.Method;
+import java.lang.reflect.Parameter;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.collectStructuredFields;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.collectTypeHierarchy;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.createRawType;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.extractAssigningConstructor;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.extractionError;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.isStructuredFieldMutable;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.resolveVariable;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.toClass;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.validateStructuredClass;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.validateStructuredFieldReadability;
+
+/**
+ * Reflection-based utility that analyzes a given {@link java.lang.reflect.Type}, method, or class to
+ * extract a (possibly nested) {@link DataType} from it.
+ */
+@Internal
+public final class DataTypeExtractor {
+
+ private final DataTypeLookup lookup;
+
+ private final String contextExplanation;
+
+ private DataTypeExtractor(DataTypeLookup lookup, String contextExplanation) {
+ this.lookup = lookup;
+ this.contextExplanation = contextExplanation;
+ }
+
+ /**
+ * Extracts a data type from a type without considering surrounding classes or templates.
+ */
+ public static DataType extractFromType(
+ DataTypeLookup lookup,
+ Type type) {
+ return extractDataTypeWithClassContext(
+ lookup,
+ DataTypeTemplate.fromDefaults(),
+ null,
+ type,
+ "");
+ }
+
+ /**
+ * Extracts a data type from a type without considering surrounding classes but templates.
+ */
+ public static DataType extractFromType(
+ DataTypeLookup lookup,
+ DataTypeTemplate template,
+ Type type) {
+ return extractDataTypeWithClassContext(
+ lookup,
+ template,
+ null,
+ type,
+ "");
+ }
+
+ /**
+ * Extracts a data type from a type variable at {@code genericPos} of {@code baseClass} using
+ * the information of the most specific type {@code contextType}.
+ */
+ public static DataType extractFromGeneric(
+ DataTypeLookup lookup,
+ Class<?> baseClass,
+ int genericPos,
+ Type contextType) {
+ final TypeVariable<?> variable = baseClass.getTypeParameters()[genericPos];
+ return extractDataTypeWithClassContext(
+ lookup,
+ DataTypeTemplate.fromDefaults(),
+ contextType,
+ variable,
+ String.format(
+ " in generic class '%s' in %s",
+ baseClass.getName(),
+ contextType.toString()));
+ }
+
+ /**
+ * Extracts a data type from a method parameter by considering surrounding classes and parameter
+ * annotation.
+ */
+ public static DataType extractFromMethodParameter(
+ DataTypeLookup lookup,
+ Class<?> baseClass,
+ Method method,
+ int paramPos) {
+ final Parameter parameter = method.getParameters()[paramPos];
+ final DataTypeHint hint = parameter.getAnnotation(DataTypeHint.class);
+ final DataTypeTemplate template;
+ if (hint != null) {
+ template = DataTypeTemplate.fromAnnotation(lookup, hint);
+ } else {
+ template = DataTypeTemplate.fromDefaults();
+ }
+ return extractDataTypeWithClassContext(
+ lookup,
+ template,
+ baseClass,
+ parameter.getParameterizedType(),
+ String.format(
+ " in parameter %d of method '%s' in class '%s'",
+ paramPos,
+ method.getName(),
+ baseClass.getName()));
+ }
+
+ /**
+ * Extracts a data type from a method return type by considering surrounding classes and method
+ * annotation.
+ */
+ public static DataType extractFromMethodOutput(
+ DataTypeLookup lookup,
+ Class<?> baseClass,
+ Method method) {
+ final DataTypeHint hint = method.getAnnotation(DataTypeHint.class);
+ final DataTypeTemplate template;
+ if (hint != null) {
+ template = DataTypeTemplate.fromAnnotation(lookup, hint);
+ } else {
+ template = DataTypeTemplate.fromDefaults();
+ }
+ return extractDataTypeWithClassContext(
+ lookup,
+ template,
+ baseClass,
+ method.getGenericReturnType(),
+ String.format(
+ " in return type of method '%s' in class '%s'",
+ method.getName(),
+ baseClass.getName()));
+ }
+
+ private static DataType extractDataTypeWithClassContext(
+ DataTypeLookup lookup,
+ DataTypeTemplate outerTemplate,
+ @Nullable Type contextType,
+ Type type,
+ String contextExplanation) {
+ final DataTypeExtractor extractor = new DataTypeExtractor(lookup, contextExplanation);
+ final List<Type> typeHierarchy;
+ if (contextType != null) {
+ typeHierarchy = collectTypeHierarchy(contextType);
+ } else {
+ typeHierarchy = Collections.emptyList();
+ }
+ return extractor.extractDataTypeOrRaw(outerTemplate, typeHierarchy, type);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private DataType extractDataTypeOrRaw(
+ DataTypeTemplate outerTemplate,
+ List<Type> typeHierarchy,
+ Type type) {
+ // best effort resolution of type variables, the resolved type can still be a variable
+ final Type resolvedType;
+ if (type instanceof TypeVariable) {
+ resolvedType = resolveVariable(typeHierarchy, (TypeVariable) type);
+ } else {
+ resolvedType = type;
+ }
+ // merge outer template with template of type itself
+ DataTypeTemplate template = outerTemplate;
+ final Class<?> clazz = toClass(resolvedType);
+ if (clazz != null) {
+ final DataTypeHint hint = clazz.getAnnotation(DataTypeHint.class);
+ if (hint != null) {
+ template = outerTemplate.mergeWithInnerAnnotation(lookup, hint);
+ }
+ }
+ // main work
+ final DataType dataType = extractDataTypeOrRawWithTemplate(template, typeHierarchy, resolvedType);
+ // final work
+ return closestBridging(dataType, clazz);
+ }
+
+ private DataType extractDataTypeOrRawWithTemplate(
+ DataTypeTemplate template,
+ List<Type> typeHierarchy,
+ Type type) {
+ // template defines a data type
+ if (template.dataType != null) {
+ return template.dataType;
+ }
+ try {
+ return extractDataTypeOrError(template, typeHierarchy, type);
+ } catch (Throwable t) {
+ // ignore the exception and just treat it as RAW type
+ final Class<?> clazz = toClass(type);
+ if (template.isAllowRawGlobally() || template.isAllowAnyPattern(clazz)) {
+ return createRawType(lookup, template.rawSerializer, clazz);
+ }
+ // forward the root cause otherwise
+ throw extractionError(
+ t,
+ "Could not extract a data type from '%s'%s. " +
+ "Please pass the required data type manually or allow RAW types.",
+ type.toString(),
+ contextExplanation);
+ }
+ }
+
+ private DataType extractDataTypeOrError(DataTypeTemplate template, List<Type> typeHierarchy, Type type) {
+ // still a type variable
+ if (type instanceof TypeVariable) {
+ throw extractionError(
+ "Unresolved type variable '%s'. A data type cannot be extracted from a type variable. " +
+ "The original content might have been erased due to Java type erasure.",
+ type.toString());
+ }
+
+ // ARRAY
+ DataType resultDataType = extractArrayType(template, typeHierarchy, type);
+ if (resultDataType != null) {
+ return resultDataType;
+ }
+
+ // skip extraction for enforced patterns early but after arrays
+ resultDataType = extractEnforcedRawType(template, type);
+ if (resultDataType != null) {
+ return resultDataType;
+ }
+
+ // PREDEFINED
+ resultDataType = extractPredefinedType(template, type);
+ if (resultDataType != null) {
+ return resultDataType;
+ }
+
+ // MAP
+ resultDataType = extractMapType(template, typeHierarchy, type);
+ if (resultDataType != null) {
+ return resultDataType;
+ }
+
+ // try interpret the type as a STRUCTURED type
+ try {
+ return extractStructuredType(template, typeHierarchy, type);
+ } catch (Throwable t) {
+ throw extractionError(
+ t,
+ "Could not extract a data type from '%s'. " +
+ "Interpreting it as a structured type was also not successful.",
+ type.toString());
+ }
+ }
+
+ private @Nullable DataType extractArrayType(
+ DataTypeTemplate template,
+ List<Type> typeHierarchy,
+ Type type) {
+ // for T[]
+ if (type instanceof GenericArrayType) {
+ final GenericArrayType genericArray = (GenericArrayType) type;
+ return DataTypes.ARRAY(
+ extractDataTypeOrRaw(template, typeHierarchy, genericArray.getGenericComponentType()));
+ }
+ // for my.custom.Pojo[][]
+ else if (type instanceof Class) {
+ final Class<?> clazz = (Class<?>) type;
+ if (clazz.isArray()) {
+ return DataTypes.ARRAY(
+ extractDataTypeOrRaw(template, typeHierarchy, clazz.getComponentType()));
+ }
+ }
+ return null;
+ }
+
+ private @Nullable DataType extractEnforcedRawType(DataTypeTemplate template, Type type) {
+ final Class<?> clazz = toClass(type);
+ if (template.isForceAnyPattern(clazz)) {
+ return createRawType(lookup, template.rawSerializer, clazz);
+ }
+ return null;
+ }
+
+ private @Nullable DataType extractPredefinedType(DataTypeTemplate template, Type type) {
+ final Class<?> clazz = toClass(type);
+ // all predefined types are representable as classes
+ if (clazz == null) {
+ return null;
+ }
+
+ // DECIMAL
+ if (clazz == BigDecimal.class) {
+ if (template.defaultDecimalPrecision != null && template.defaultDecimalScale != null) {
+ return DataTypes.DECIMAL(template.defaultDecimalPrecision, template.defaultDecimalScale);
+ } else if (template.defaultDecimalPrecision != null) {
+ return DataTypes.DECIMAL(template.defaultDecimalPrecision, 0);
+ }
+ throw extractionError("Values of '%s' need fixed precision and scale.", BigDecimal.class.getName());
+ }
+
+ // TIME
+ else if (clazz == java.sql.Time.class || clazz == java.time.LocalTime.class) {
+ if (template.defaultSecondPrecision != null) {
+ return DataTypes.TIME(template.defaultSecondPrecision)
+ .bridgedTo(clazz);
+ }
+ }
+
+ // TIMESTAMP
+ else if (clazz == java.sql.Timestamp.class || clazz == java.time.LocalDateTime.class) {
+ if (template.defaultSecondPrecision != null) {
+ return DataTypes.TIMESTAMP(template.defaultSecondPrecision)
+ .bridgedTo(clazz);
+ }
+ }
+
+ // TIMESTAMP WITH TIME ZONE
+ else if (clazz == java.time.OffsetDateTime.class) {
+ if (template.defaultSecondPrecision != null) {
+ return DataTypes.TIMESTAMP_WITH_TIME_ZONE(template.defaultSecondPrecision);
+ }
+ }
+
+ // TIMESTAMP WITH LOCAL TIME ZONE
+ else if (clazz == java.time.Instant.class) {
+ if (template.defaultSecondPrecision != null) {
+ return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(template.defaultSecondPrecision);
+ }
+ }
+
+ // INTERVAL SECOND
+ else if (clazz == java.time.Duration.class) {
+ if (template.defaultSecondPrecision != null) {
+ return DataTypes.INTERVAL(DataTypes.SECOND(template.defaultSecondPrecision));
+ }
+ }
+
+ // INTERVAL YEAR TO MONTH
+ else if (clazz == java.time.Period.class) {
+ if (template.defaultYearPrecision != null && template.defaultYearPrecision == 0) {
+ return DataTypes.INTERVAL(DataTypes.MONTH());
+ } else if (template.defaultYearPrecision != null) {
+ return DataTypes.INTERVAL(DataTypes.YEAR(template.defaultYearPrecision), DataTypes.MONTH());
+ }
+ }
+
+ return ClassDataTypeConverter.extractDataType(clazz).orElse(null);
+ }
+
+ private @Nullable DataType extractMapType(DataTypeTemplate template, List<Type> typeHierarchy, Type type) {
+ final Class<?> clazz = toClass(type);
+ if (clazz != Map.class) {
+ return null;
+ }
+ if (!(type instanceof ParameterizedType)) {
+ throw extractionError("Raw map type needs generic parameters.");
+ }
+ final ParameterizedType parameterizedType = (ParameterizedType) type;
+ final DataType key = extractDataTypeOrRaw(
+ template,
+ typeHierarchy,
+ parameterizedType.getActualTypeArguments()[0]);
+ final DataType value = extractDataTypeOrRaw(
+ template,
+ typeHierarchy,
+ parameterizedType.getActualTypeArguments()[1]);
+ return DataTypes.MAP(key, value);
+ }
+
+ private DataType extractStructuredType(DataTypeTemplate template, List<Type> typeHierarchy, Type type) {
+ final Class<?> clazz = toClass(type);
+ if (clazz == null) {
+ throw extractionError("Not a class type.");
+ }
+
+ validateStructuredClass(clazz);
+
+ final List<Field> fields = collectStructuredFields(clazz);
+
+ if (fields.isEmpty()) {
+ throw extractionError(
+ "Class '%s' has no fields.",
+ clazz.getName());
+ }
+
+ // if not all fields are mutable, a default constructor is not enough
+ final boolean allFieldsMutable = fields.stream().allMatch(f -> {
+ validateStructuredFieldReadability(clazz, f);
+ return isStructuredFieldMutable(clazz, f);
+ });
+
+ final ExtractionUtils.AssigningConstructor constructor = extractAssigningConstructor(clazz, fields);
+ if (!allFieldsMutable && constructor == null) {
+ throw extractionError(
+ "Class '%s' has immutable fields and thus requires a constructor that is publicly " +
+ "accessible and assigns all fields: %s",
+ clazz.getName(),
+ fields.stream().map(Field::getName).collect(Collectors.joining(", ")));
+ }
+
+ final Map<String, DataType> fieldDataTypes = extractStructuredTypeFields(
+ template,
+ typeHierarchy,
+ type,
+ fields);
+
+ final StructuredType.Builder builder = StructuredType.newBuilder(clazz);
+ builder.attributes(createStructuredTypeAttributes(constructor, fieldDataTypes));
+ builder.setFinal(true); // anonymous structured types should not allow inheritance
+ builder.setInstantiable(true);
+ return new FieldsDataType(builder.build(), clazz, fieldDataTypes);
+ }
+
+ private Map<String, DataType> extractStructuredTypeFields(
+ DataTypeTemplate template,
+ List<Type> typeHierarchy,
+ Type type,
+ List<Field> fields) {
+ final Map<String, DataType> fieldDataTypes = new HashMap<>();
+ final List<Type> structuredTypeHierarchy = collectTypeHierarchy(type);
+ for (Field field : fields) {
+ try {
+ final Type fieldType = field.getGenericType();
+ final List<Type> fieldTypeHierarchy = new ArrayList<>();
+ // hierarchy until structured type
+ fieldTypeHierarchy.addAll(typeHierarchy);
+ // hierarchy of structured type
+ fieldTypeHierarchy.addAll(structuredTypeHierarchy);
+ final DataTypeTemplate fieldTemplate = mergeFieldTemplate(lookup, field, template);
+ final DataType fieldDataType = extractDataTypeOrRaw(fieldTemplate, fieldTypeHierarchy, fieldType);
+ fieldDataTypes.put(field.getName(), fieldDataType);
+ } catch (Throwable t) {
+ throw extractionError(
+ t,
+ "Error in field '%s' of class '%s'.",
+ field.getName(),
+ field.getDeclaringClass().getName());
+ }
+ }
+ return fieldDataTypes;
+ }
+
+ private List<StructuredAttribute> createStructuredTypeAttributes(
+ ExtractionUtils.AssigningConstructor constructor,
+ Map<String, DataType> fieldDataTypes) {
+ return Optional.ofNullable(constructor)
+ .map(c -> {
+ // field order is defined by assigning constructor
+ return c.parameterNames.stream();
+ })
+ .orElseGet(() -> {
+ // field order is sorted
+ return fieldDataTypes.keySet().stream().sorted();
+ })
+ .map(name -> {
+ final LogicalType logicalType = fieldDataTypes.get(name).getLogicalType();
+ return new StructuredAttribute(name, logicalType);
+ })
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Merges the template of a structured type with a possibly more specific field annotation.
+ */
+ private DataTypeTemplate mergeFieldTemplate(DataTypeLookup lookup, Field field, DataTypeTemplate structuredTemplate) {
+ final DataTypeHint hint = field.getAnnotation(DataTypeHint.class);
+ if (hint == null) {
+ return structuredTemplate.copyWithoutDataType();
+ }
+ return structuredTemplate.mergeWithInnerAnnotation(lookup, hint);
+ }
+
+ /**
+ * Use closest class for data type if possible. Even though a hint might have provided some data
+ * type, in many cases, the conversion class can be enriched with the extraction type itself.
+ */
+ @SuppressWarnings("unchecked")
+ private DataType closestBridging(DataType dataType, @Nullable Class<?> clazz) {
+ // no context class or conversion class is already more specific than context class
+ if (clazz == null || clazz.isAssignableFrom(dataType.getConversionClass())) {
+ return dataType;
+ }
+ final LogicalType logicalType = dataType.getLogicalType();
+ final boolean supportsConversion = logicalType.supportsInputConversion(clazz) ||
+ logicalType.supportsOutputConversion(clazz);
+ if (supportsConversion && logicalType instanceof RawType) {
+ return DataTypes.RAW(clazz, ((RawType) logicalType).getTypeSerializer());
+ } else if (supportsConversion) {
+ return dataType.bridgedTo(clazz);
+ }
+ return dataType;
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java
new file mode 100644
index 0000000..5a73a24
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java
@@ -0,0 +1,384 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction.utils;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ExtractionVersion;
+import org.apache.flink.table.annotation.HintFlag;
+import org.apache.flink.table.annotation.InputGroup;
+import org.apache.flink.table.catalog.DataTypeLookup;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.extraction.DataTypeExtractor;
+import org.apache.flink.table.types.inference.ArgumentTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeValidators;
+import org.apache.flink.table.types.inference.TypeStrategies;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.function.Function;
+
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.createRawType;
+import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.extractionError;
+
+/**
+ * Internal representation of a {@link DataTypeHint} and template for creating a single {@link DataType}
+ * or a {@link InputTypeValidator} for groups of {@link DataType}s.
+ *
+ * <p>All parameters of a template are optional. An empty annotation results in a template where all
+ * members are {@code null}.
+ */
+@Internal
+public final class DataTypeTemplate {
+
+ private static final DataTypeHint DEFAULT_ANNOTATION = getDefaultAnnotation();
+
+ private static final String RAW_TYPE_NAME = "RAW";
+
+ public final @Nullable DataType dataType;
+
+ public final @Nullable Class<? extends TypeSerializer<?>> rawSerializer;
+
+ public final @Nullable InputGroup inputGroup;
+
+ public final @Nullable ExtractionVersion version;
+
+ public final @Nullable Boolean allowRawGlobally;
+
+ public final @Nullable String[] allowRawPattern;
+
+ public final @Nullable String[] forceRawPattern;
+
+ public final @Nullable Integer defaultDecimalPrecision;
+
+ public final @Nullable Integer defaultDecimalScale;
+
+ public final @Nullable Integer defaultYearPrecision;
+
+ public final @Nullable Integer defaultSecondPrecision;
+
+ private DataTypeTemplate(
+ @Nullable DataType dataType,
+ @Nullable Class<? extends TypeSerializer<?>> rawSerializer,
+ @Nullable InputGroup inputGroup,
+ @Nullable ExtractionVersion version,
+ @Nullable Boolean allowRawGlobally,
+ @Nullable String[] allowRawPattern,
+ @Nullable String[] forceRawPattern,
+ @Nullable Integer defaultDecimalPrecision,
+ @Nullable Integer defaultDecimalScale,
+ @Nullable Integer defaultYearPrecision,
+ @Nullable Integer defaultSecondPrecision) {
+ this.dataType = dataType;
+ this.rawSerializer = rawSerializer;
+ this.inputGroup = inputGroup;
+ this.version = version;
+ this.allowRawGlobally = allowRawGlobally;
+ this.allowRawPattern = allowRawPattern;
+ this.forceRawPattern = forceRawPattern;
+ this.defaultDecimalPrecision = defaultDecimalPrecision;
+ this.defaultDecimalScale = defaultDecimalScale;
+ this.defaultYearPrecision = defaultYearPrecision;
+ this.defaultSecondPrecision = defaultSecondPrecision;
+ }
+
+ /**
+ * Creates an instance from the given {@link DataTypeHint}. Resolves an explicitly defined data type
+ * if {@link DataTypeHint#value()} and/or {@link DataTypeHint#bridgedTo()} are defined.
+ */
+ public static DataTypeTemplate fromAnnotation(DataTypeLookup lookup, DataTypeHint hint) {
+ final String typeName = defaultAsNull(hint, DataTypeHint::value);
+ final Class<?> conversionClass = defaultAsNull(hint, DataTypeHint::bridgedTo);
+ if (typeName != null || conversionClass != null) {
+ // chicken and egg problem
+ // a template can contain a data type but in order to extract a data type we might need a template
+ final DataTypeTemplate extractionTemplate = fromAnnotation(hint, null);
+ return fromAnnotation(hint, extractDataType(lookup, typeName, conversionClass, extractionTemplate));
+ }
+ return fromAnnotation(hint, null);
+ }
+
+ /**
+ * Creates an instance from the given {@link DataTypeHint} with a resolved data type if available.
+ */
+ public static DataTypeTemplate fromAnnotation(DataTypeHint hint, @Nullable DataType dataType) {
+ return new DataTypeTemplate(
+ dataType,
+ defaultAsNull(hint, DataTypeHint::rawSerializer),
+ defaultAsNull(hint, DataTypeHint::inputGroup),
+ defaultAsNull(hint, DataTypeHint::version),
+ hintFlagToBoolean(defaultAsNull(hint, DataTypeHint::allowRawGlobally)),
+ defaultAsNull(hint, DataTypeHint::allowRawPattern),
+ defaultAsNull(hint, DataTypeHint::forceRawPattern),
+ defaultAsNull(hint, DataTypeHint::defaultDecimalPrecision),
+ defaultAsNull(hint, DataTypeHint::defaultDecimalScale),
+ defaultAsNull(hint, DataTypeHint::defaultYearPrecision),
+ defaultAsNull(hint, DataTypeHint::defaultSecondPrecision)
+ );
+ }
+
+ /**
+ * Creates an instance with no parameter content.
+ */
+ public static DataTypeTemplate fromDefaults() {
+ return new DataTypeTemplate(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+ }
+
+ /**
+ * Copies this template but removes the explicit data type (if available).
+ */
+ public DataTypeTemplate copyWithoutDataType() {
+ return new DataTypeTemplate(
+ null,
+ rawSerializer,
+ inputGroup,
+ version,
+ allowRawGlobally,
+ allowRawPattern,
+ forceRawPattern,
+ defaultDecimalPrecision,
+ defaultDecimalScale,
+ defaultYearPrecision,
+ defaultSecondPrecision
+ );
+ }
+
+ /**
+ * Merges this template with an inner annotation. The inner annotation has highest precedence
+ * and definitely determines the explicit data type (if available).
+ */
+ public DataTypeTemplate mergeWithInnerAnnotation(DataTypeLookup lookup, DataTypeHint hint) {
+ final DataTypeTemplate otherTemplate = fromAnnotation(lookup, hint);
+ return new DataTypeTemplate(
+ otherTemplate.dataType,
+ rightValueIfNotNull(rawSerializer, otherTemplate.rawSerializer),
+ rightValueIfNotNull(inputGroup, otherTemplate.inputGroup),
+ rightValueIfNotNull(version, otherTemplate.version),
+ rightValueIfNotNull(allowRawGlobally, otherTemplate.allowRawGlobally),
+ rightValueIfNotNull(allowRawPattern, otherTemplate.allowRawPattern),
+ rightValueIfNotNull(forceRawPattern, otherTemplate.forceRawPattern),
+ rightValueIfNotNull(defaultDecimalPrecision, otherTemplate.defaultDecimalPrecision),
+ rightValueIfNotNull(defaultDecimalScale, otherTemplate.defaultDecimalScale),
+ rightValueIfNotNull(defaultYearPrecision, otherTemplate.defaultYearPrecision),
+ rightValueIfNotNull(defaultSecondPrecision, otherTemplate.defaultSecondPrecision)
+ );
+ }
+
+ /**
+ * Whether this template defines an explicit data type.
+ */
+ public boolean hasDataTypeDefinition() {
+ return dataType != null;
+ }
+
+ /**
+ * Whether this template defines a group of data types for an input argument.
+ */
+ public boolean hasInputGroupDefinition() {
+ return inputGroup != null && inputGroup != InputGroup.UNKNOWN;
+ }
+
+ /**
+ * Converts this template into an {@link ArgumentTypeValidator}.
+ */
+ public ArgumentTypeValidator toArgumentTypeValidator() {
+ // data type
+ if (hasDataTypeDefinition()) {
+ return InputTypeValidators.explicit(dataType);
+ }
+ // input group
+ else if (hasInputGroupDefinition()) {
+ if (inputGroup == InputGroup.ANY) {
+ return InputTypeValidators.ANY;
+ }
+ }
+ throw ExtractionUtils.extractionError(
+ "Data type hint does neither specify an explicit data type nor an input group.");
+ }
+
+ /**
+ * Converts this template into a {@link TypeStrategy}.
+ */
+ public TypeStrategy toTypeStrategy() {
+ if (hasDataTypeDefinition()) {
+ return TypeStrategies.explicit(dataType);
+ }
+ throw ExtractionUtils.extractionError(
+ "Data type hint does not specify an explicit data type.");
+ }
+
+ /**
+ * Returns whether RAW types are allowed everywhere.
+ */
+ public boolean isAllowRawGlobally() {
+ return allowRawGlobally != null && allowRawGlobally;
+ }
+
+ /**
+ * Returns whether the given class is eligible for being treated as RAW type.
+ */
+ public boolean isAllowAnyPattern(@Nullable Class<?> clazz) {
+ if (allowRawPattern == null || clazz == null) {
+ return false;
+ }
+ final String className = clazz.getName();
+ for (String pattern : allowRawPattern) {
+ if (className.startsWith(pattern)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns whether the given class must be treated as RAW type.
+ */
+ public boolean isForceAnyPattern(@Nullable Class<?> clazz) {
+ if (forceRawPattern == null || clazz == null) {
+ return false;
+ }
+ final String className = clazz.getName();
+ for (String pattern : forceRawPattern) {
+ if (className.startsWith(pattern)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DataTypeTemplate that = (DataTypeTemplate) o;
+ return Objects.equals(dataType, that.dataType) &&
+ Objects.equals(rawSerializer, that.rawSerializer) &&
+ Objects.equals(inputGroup, that.inputGroup) &&
+ version == that.version &&
+ Objects.equals(allowRawGlobally, that.allowRawGlobally) &&
+ Arrays.equals(allowRawPattern, that.allowRawPattern) &&
+ Arrays.equals(forceRawPattern, that.forceRawPattern) &&
+ Objects.equals(defaultDecimalPrecision, that.defaultDecimalPrecision) &&
+ Objects.equals(defaultDecimalScale, that.defaultDecimalScale) &&
+ Objects.equals(defaultYearPrecision, that.defaultYearPrecision) &&
+ Objects.equals(defaultSecondPrecision, that.defaultSecondPrecision);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Objects.hash(
+ dataType,
+ rawSerializer,
+ inputGroup,
+ version,
+ allowRawGlobally,
+ defaultDecimalPrecision,
+ defaultDecimalScale,
+ defaultYearPrecision,
+ defaultSecondPrecision);
+ result = 31 * result + Arrays.hashCode(allowRawPattern);
+ result = 31 * result + Arrays.hashCode(forceRawPattern);
+ return result;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @DataTypeHint
+ private static class DefaultAnnotationHelper {
+ // no implementation
+ }
+
+ private static DataTypeHint getDefaultAnnotation() {
+ return DefaultAnnotationHelper.class.getAnnotation(DataTypeHint.class);
+ }
+
+ private static <T> T defaultAsNull(DataTypeHint hint, Function<DataTypeHint, T> accessor) {
+ final T defaultValue = accessor.apply(DEFAULT_ANNOTATION);
+ final T actualValue = accessor.apply(hint);
+ if (Objects.deepEquals(defaultValue, actualValue)) {
+ return null;
+ }
+ return actualValue;
+ }
+
+ private static <T> T rightValueIfNotNull(T l, T r) {
+ if (r != null) {
+ return r;
+ }
+ return l;
+ }
+
+ private static Boolean hintFlagToBoolean(HintFlag flag) {
+ if (flag == null) {
+ return null;
+ }
+ return flag == HintFlag.TRUE;
+ }
+
+ private static DataType extractDataType(
+ DataTypeLookup lookup,
+ @Nullable String typeName,
+ @Nullable Class<?> conversionClass,
+ DataTypeTemplate template) {
+ // explicit data type
+ if (typeName != null) {
+ // RAW type
+ if (typeName.equals(RAW_TYPE_NAME)) {
+ return createRawType(lookup, template.rawSerializer, conversionClass);
+ }
+ // regular type that must be resolvable
+ return lookup.lookupDataType(typeName)
+ .map(dataType -> {
+ if (conversionClass != null) {
+ return dataType.bridgedTo(conversionClass);
+ }
+ return dataType;
+ })
+ .orElseThrow(() -> extractionError("Could not resolve type with name '%s'.", typeName));
+ }
+ // extracted data type
+ else if (conversionClass != null) {
+ return DataTypeExtractor.extractFromType(lookup, template, conversionClass);
+ }
+ throw ExtractionUtils.extractionError(
+ "Data type hint does neither specify an explicit data type or conversion class " +
+ "from which a data type could be extracted.");
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/ExtractionUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/ExtractionUtils.java
new file mode 100644
index 0000000..7e5a94f
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/ExtractionUtils.java
@@ -0,0 +1,571 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction.utils;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.DataTypeLookup;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.StructuredType;
+
+import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassReader;
+import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassVisitor;
+import org.apache.flink.shaded.asm7.org.objectweb.asm.Label;
+import org.apache.flink.shaded.asm7.org.objectweb.asm.MethodVisitor;
+import org.apache.flink.shaded.asm7.org.objectweb.asm.Opcodes;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.lang.reflect.Parameter;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor;
+
+/**
+ * Utilities for performing reflection tasks.
+ */
+public final class ExtractionUtils {
+
+ /**
+ * Helper method for creating consistent exceptions during extraction.
+ */
+ public static ValidationException extractionError(String message, Object... args) {
+ return extractionError(null, message, args);
+ }
+
+ /**
+ * Helper method for creating consistent exceptions during extraction.
+ */
+ public static ValidationException extractionError(Throwable cause, String message, Object... args) {
+ return new ValidationException(
+ String.format(
+ message,
+ args
+ ),
+ cause
+ );
+ }
+
+ /**
+ * Collects the partially ordered type hierarchy (i.e. all involved super classes and super
+ * interfaces) of the given type.
+ */
+ public static List<Type> collectTypeHierarchy(Type type) {
+ Type currentType = type;
+ Class<?> currentClass = toClass(type);
+ final List<Type> typeHierarchy = new ArrayList<>();
+ while (currentClass != null) {
+ // collect type
+ typeHierarchy.add(currentType);
+ // collect super interfaces
+ for (Type genericInterface : currentClass.getGenericInterfaces()) {
+ final Class<?> interfaceClass = toClass(genericInterface);
+ if (interfaceClass != null) {
+ typeHierarchy.addAll(collectTypeHierarchy(genericInterface));
+ }
+ }
+ currentType = currentClass.getGenericSuperclass();
+ currentClass = toClass(currentType);
+ }
+ return typeHierarchy;
+ }
+
+ /**
+ * Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise.
+ */
+ public static @Nullable Class<?> toClass(Type type) {
+ if (type instanceof Class) {
+ return (Class<?>) type;
+ } else if (type instanceof ParameterizedType) {
+ // this is always a class
+ return (Class<?>) ((ParameterizedType) type).getRawType();
+ }
+ // unsupported: generic arrays, type variables, wildcard types
+ return null;
+ }
+
+ /**
+ * Creates a raw data type.
+ */
+ @SuppressWarnings("unchecked")
+ public static DataType createRawType(
+ DataTypeLookup lookup,
+ @Nullable Class<? extends TypeSerializer<?>> rawSerializer,
+ @Nullable Class<?> conversionClass) {
+ if (rawSerializer != null) {
+ return DataTypes.RAW((Class) createConversionClass(conversionClass), instantiateRawSerializer(rawSerializer));
+ }
+ return lookup.resolveRawDataType(createConversionClass(conversionClass));
+ }
+
+ private static Class<?> createConversionClass(@Nullable Class<?> conversionClass) {
+ if (conversionClass != null) {
+ return conversionClass;
+ }
+ return Object.class;
+ }
+
+ private static TypeSerializer<?> instantiateRawSerializer(Class<? extends TypeSerializer<?>> rawSerializer) {
+ try {
+ return rawSerializer.newInstance();
+ } catch (Exception e) {
+ throw extractionError(
+ e,
+ "Cannot instantiate type serializer '%s' for RAW type. " +
+ "Make sure the class is publicly accessible and has a default constructor.",
+ rawSerializer.getName()
+ );
+ }
+ }
+
+ /**
+ * Resolves a {@link TypeVariable} using the given type hierarchy if possible.
+ */
+ public static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) {
+ // iterate through hierarchy from top to bottom until type variable gets a non-variable assigned
+ for (int i = typeHierarchy.size() - 1; i >= 0; i--) {
+ final Type currentType = typeHierarchy.get(i);
+
+ if (currentType instanceof ParameterizedType) {
+ final Type resolvedType = resolveVariableInParameterizedType(
+ variable,
+ (ParameterizedType) currentType);
+ if (resolvedType instanceof TypeVariable) {
+ // follow type variables transitively
+ variable = (TypeVariable<?>) resolvedType;
+ } else if (resolvedType != null) {
+ return resolvedType;
+ }
+ }
+ }
+ // unresolved variable
+ return variable;
+ }
+
+ private static @Nullable Type resolveVariableInParameterizedType(
+ TypeVariable<?> variable,
+ ParameterizedType currentType) {
+ final Class<?> currentRaw = (Class<?>) currentType.getRawType();
+ final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters();
+ // search for matching type variable
+ for (int paramPos = 0; paramPos < currentVariables.length; paramPos++) {
+ if (typeVariableEquals(variable, currentVariables[paramPos])) {
+ return currentType.getActualTypeArguments()[paramPos];
+ }
+ }
+ return null;
+ }
+
+ private static boolean typeVariableEquals(TypeVariable<?> variable, TypeVariable<?> currentVariable) {
+ return currentVariable.getGenericDeclaration().equals(variable.getGenericDeclaration()) &&
+ currentVariable.getName().equals(variable.getName());
+ }
+
+ /**
+ * Validates the characteristics of a class for a {@link StructuredType} such as accessibility.
+ */
+ public static void validateStructuredClass(Class<?> clazz) {
+ final int m = clazz.getModifiers();
+ if (Modifier.isAbstract(m)) {
+ throw extractionError("Class '%s' must not be abstract.", clazz.getName());
+ }
+ if (!Modifier.isPublic(m)) {
+ throw extractionError("Class '%s' is not public.", clazz.getName());
+ }
+ if (clazz.getEnclosingClass() != null &&
+ (clazz.getDeclaringClass() == null || !Modifier.isStatic(m))) {
+ throw extractionError("Class '%s' is a not a static, globally accessible class.", clazz.getName());
+ }
+ }
+
+ /**
+ * Returns the fields of a class for a {@link StructuredType}.
+ */
+ public static List<Field> collectStructuredFields(Class<?> clazz) {
+ final List<Field> fields = new ArrayList<>();
+ while (clazz != Object.class) {
+ final Field[] declaredFields = clazz.getDeclaredFields();
+ Stream.of(declaredFields)
+ .filter(field -> {
+ final int m = field.getModifiers();
+ return !Modifier.isStatic(m) && !Modifier.isTransient(m);
+ })
+ .forEach(fields::add);
+ clazz = clazz.getSuperclass();
+ }
+ return fields;
+ }
+
+ /**
+ * Validates if a field is properly readable either directly or through a getter.
+ */
+ public static void validateStructuredFieldReadability(Class<?> clazz, Field field) {
+ final int m = field.getModifiers();
+
+ // field is accessible
+ if (Modifier.isPublic(m)) {
+ return;
+ }
+
+ // field needs a getter
+ if (!hasStructuredFieldGetter(clazz, field)) {
+ throw extractionError(
+ "Field '%s' of class '%s' is neither publicly accessible nor does it have " +
+ "a corresponding getter method.",
+ field.getName(),
+ clazz.getName());
+ }
+ }
+
+ /**
+ * Checks if a field is mutable or immutable. Returns {@code true} if the field is properly
+ * mutable. Returns {@code false} if it is properly immutable.
+ */
+ public static boolean isStructuredFieldMutable(Class<?> clazz, Field field) {
+ final int m = field.getModifiers();
+
+ // field is immutable
+ if (Modifier.isFinal(m)) {
+ return false;
+ }
+ // field is directly mutable
+ if (Modifier.isPublic(m)) {
+ return true;
+ }
+ // field has setters by which it is mutable
+ if (hasFieldSetter(clazz, field)) {
+ return true;
+ }
+
+ throw extractionError(
+ "Field '%s' of class '%s' is mutable but is neither publicly accessible nor does it have " +
+ "a corresponding setter method.",
+ field.getName(),
+ clazz.getName());
+ }
+
+ /**
+ * Checks for a field setters. The logic is as broad as possible to support both Java and Scala
+ * in different flavors.
+ */
+ public static boolean hasFieldSetter(Class<?> clazz, Field field) {
+ final String normalizedFieldName = field.getName().toUpperCase();
+
+ final List<Method> methods = collectStructuredMethods(clazz);
+ for (Method method : methods) {
+
+ // check name:
+ // set<Name>(type)
+ // <Name>(type)
+ // <Name>_$eq(type) for Scala
+ final String normalizedMethodName = method.getName().toUpperCase();
+ final boolean hasName = normalizedMethodName.equals("SET" + normalizedFieldName) ||
+ normalizedMethodName.equals(normalizedFieldName) ||
+ normalizedMethodName.equals(normalizedFieldName + "_$EQ");
+ if (!hasName) {
+ continue;
+ }
+
+ // check return type:
+ // void or the declaring class
+ final Class<?> returnType = method.getReturnType();
+ final boolean hasReturnType = returnType == Void.TYPE || returnType == clazz;
+ if (!hasReturnType) {
+ continue;
+ }
+
+ // check parameters:
+ // one parameter that has the same (or primitive) type of the field
+ final boolean hasParameter = method.getParameterCount() == 1 &&
+ (method.getGenericParameterTypes()[0].equals(field.getGenericType()) ||
+ boxPrimitive(method.getGenericParameterTypes()[0]).equals(field.getGenericType()));
+ if (!hasParameter) {
+ continue;
+ }
+
+ // matching setter found
+ return true;
+ }
+
+ // no setter found
+ return false;
+ }
+
+ private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>();
+ static {
+ primitiveWrapperMap.put(Boolean.TYPE, Boolean.class);
+ primitiveWrapperMap.put(Byte.TYPE, Byte.class);
+ primitiveWrapperMap.put(Character.TYPE, Character.class);
+ primitiveWrapperMap.put(Short.TYPE, Short.class);
+ primitiveWrapperMap.put(Integer.TYPE, Integer.class);
+ primitiveWrapperMap.put(Long.TYPE, Long.class);
+ primitiveWrapperMap.put(Double.TYPE, Double.class);
+ primitiveWrapperMap.put(Float.TYPE, Float.class);
+ }
+
+ /**
+ * Returns the boxed type of a primitive type.
+ */
+ public static Type boxPrimitive(Type type) {
+ if (type instanceof Class && ((Class) type).isPrimitive()) {
+ return primitiveWrapperMap.get(type);
+ }
+ return type;
+ }
+
+ /**
+ * Checks for a field getter. The logic is as broad as possible to support both Java and Scala
+ * in different flavors.
+ */
+ public static boolean hasStructuredFieldGetter(Class<?> clazz, Field field) {
+ final String normalizedFieldName = field.getName().toUpperCase();
+
+ final List<Method> methods = collectStructuredMethods(clazz);
+ for (Method method : methods) {
+ // check name:
+ // get<Name>()
+ // is<Name>()
+ // <Name>() for Scala
+ final String normalizedMethodName = method.getName().toUpperCase();
+ final boolean hasName = normalizedMethodName.equals("GET" + normalizedFieldName) ||
+ normalizedMethodName.equals("IS" + normalizedFieldName) ||
+ normalizedMethodName.equals(normalizedFieldName);
+ if (!hasName) {
+ continue;
+ }
+
+ // check return type:
+ // equal to field type
+ final Type returnType = method.getGenericReturnType();
+ final boolean hasReturnType = returnType.equals(field.getGenericType());
+ if (!hasReturnType) {
+ continue;
+ }
+
+ // check parameters:
+ // no parameters
+ final boolean hasNoParameters = method.getParameterCount() == 0;
+ if (!hasNoParameters) {
+ continue;
+ }
+
+ // matching getter found
+ return true;
+ }
+
+ // no getter found
+ return false;
+ }
+
+ /**
+ * Collects all methods that qualify as methods of a {@link StructuredType}.
+ */
+ public static List<Method> collectStructuredMethods(Class<?> clazz) {
+ final List<Method> methods = new ArrayList<>();
+ while (clazz != Object.class) {
+ final Method[] declaredMethods = clazz.getDeclaredMethods();
+ Stream.of(declaredMethods)
+ .filter(field -> {
+ final int m = field.getModifiers();
+ return Modifier.isPublic(m) && !Modifier.isNative(m) && !Modifier.isAbstract(m);
+ })
+ .forEach(methods::add);
+ clazz = clazz.getSuperclass();
+ }
+ return methods;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Assignable Constructor Utilities
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Result of the extraction in {@link #extractAssigningConstructor(Class, List)}.
+ */
+ public static class AssigningConstructor {
+ public final Constructor<?> constructor;
+ public final List<String> parameterNames;
+
+ private AssigningConstructor(Constructor<?> constructor, List<String> parameterNames) {
+ this.constructor = constructor;
+ this.parameterNames = parameterNames;
+ }
+ }
+
+ /**
+ * Checks whether the given constructor takes all of the given fields with matching (possibly
+ * primitive) type and name. An assigning constructor can define the order of fields.
+ */
+ public static @Nullable AssigningConstructor extractAssigningConstructor(
+ Class<?> clazz,
+ List<Field> fields) {
+ AssigningConstructor foundConstructor = null;
+ for (Constructor<?> constructor : clazz.getDeclaredConstructors()) {
+ final boolean qualifyingConstructor = Modifier.isPublic(constructor.getModifiers()) &&
+ constructor.getParameterTypes().length == fields.size();
+ if (!qualifyingConstructor) {
+ continue;
+ }
+ final List<String> parameterNames = extractConstructorParameterNames(clazz, constructor, fields);
+ if (parameterNames != null) {
+ if (foundConstructor != null) {
+ throw extractionError(
+ "Multiple constructors found that assign all fields for class '%s'.",
+ clazz.getName());
+ }
+ foundConstructor = new AssigningConstructor(constructor, parameterNames);
+ }
+ }
+ return foundConstructor;
+ }
+
+ /**
+ * Extracts ordered parameter names from a constructor that takes all of the given fields with
+ * matching (possibly primitive) type and name.
+ */
+ private static @Nullable List<String> extractConstructorParameterNames(
+ Class<?> clazz,
+ Constructor<?> constructor,
+ List<Field> fields) {
+ final Type[] parameterTypes = constructor.getGenericParameterTypes();
+
+ // by default parameter names are "arg0, arg1, arg2, ..." if compiler flag is not set
+ // so we need to extract them manually if possible
+ List<String> parameterNames = Stream.of(constructor.getParameters())
+ .map(Parameter::getName)
+ .collect(Collectors.toList());
+ if (parameterNames.stream().allMatch(n -> n.startsWith("arg"))) {
+ final ParameterExtractor extractor = new ParameterExtractor(constructor);
+ getClassReader(clazz).accept(extractor, 0);
+
+ final List<String> extractedNames = extractor.getParameterNames();
+ if (extractedNames.size() == 0 || !extractedNames.get(0).equals("this")) {
+ return null;
+ }
+ // remove "this" and additional local variables
+ // select less names if class file has not the required information
+ parameterNames = extractedNames.subList(1, Math.min(fields.size() + 1, extractedNames.size()));
+ }
+
+ if (parameterNames.size() != fields.size()) {
+ return null;
+ }
+
+ final Map<String, Type> fieldMap = fields.stream()
+ .collect(Collectors.toMap(Field::getName, Field::getGenericType));
+
+ // check that all fields are represented in the parameters of the constructor
+ for (int i = 0; i < parameterNames.size(); i++) {
+ final String parameterName = parameterNames.get(i);
+ final Type fieldType = fieldMap.get(parameterName); // might be null
+ final Type parameterType = parameterTypes[i];
+ // we are tolerant here because frameworks such as Avro accept a boxed type even though
+ // the field is primitive
+ if (!boxPrimitive(parameterType).equals(boxPrimitive(fieldType))) {
+ return null;
+ }
+ }
+
+ return parameterNames;
+ }
+
+ private static ClassReader getClassReader(Class<?> cls) {
+ final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
+ try {
+ return new ClassReader(cls.getResourceAsStream(className));
+ } catch (IOException e) {
+ throw new IllegalStateException("Could not instantiate ClassReader.", e);
+ }
+ }
+
+ /**
+ * Extracts the parameter names and descriptors from a constructor. Assuming the existence of a
+ * local variable table.
+ *
+ * <p>For example:
+ * <pre>
+ * {@code
+ * public WC(java.lang.String arg0, long arg1) { // <init> //(Ljava/lang/String;J)V
+ * <localVar:index=0 , name=this , desc=Lorg/apache/flink/WC;, sig=null, start=L1, end=L2>
+ * <localVar:index=1 , name=word , desc=Ljava/lang/String;, sig=null, start=L1, end=L2>
+ * <localVar:index=2 , name=frequency , desc=J, sig=null, start=L1, end=L2>
+ * <localVar:index=2 , name=otherLocal , desc=J, sig=null, start=L1, end=L2>
+ * <localVar:index=2 , name=otherLocal2 , desc=J, sig=null, start=L1, end=L2>
+ * }
+ * }
+ * </pre>
+ */
+ private static class ParameterExtractor extends ClassVisitor {
+
+ private final String constructorDescriptor;
+
+ private final List<String> parameterNames = new ArrayList<>();
+
+ public ParameterExtractor(Constructor constructor) {
+ super(Opcodes.ASM6);
+ constructorDescriptor = getConstructorDescriptor(constructor);
+ }
+
+ public List<String> getParameterNames() {
+ return parameterNames;
+ }
+
+ @Override
+ public MethodVisitor visitMethod(
+ int access,
+ String name,
+ String descriptor,
+ String signature,
+ String[] exceptions) {
+ if (descriptor.equals(constructorDescriptor)) {
+ return new MethodVisitor(Opcodes.ASM6) {
+ @Override
+ public void visitLocalVariable(
+ String name,
+ String descriptor,
+ String signature,
+ Label start,
+ Label end,
+ int index) {
+ parameterNames.add(name);
+ }
+ };
+ }
+ return super.visitMethod(access, name, descriptor, signature, exceptions);
+ }
+ }
+
+ private ExtractionUtils() {
+ // no instantiation
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
index 59370f9..f292dca 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
@@ -41,7 +41,7 @@ public final class ClassDataTypeConverter {
private static final Map<String, DataType> defaultDataTypes = new HashMap<>();
static {
// NOTE: this list explicitly excludes data types that need further parameters
- // exclusions: DECIMAL, INTERVAL YEAR TO MONTH, MAP, MULTISET, ROW, NULL, ANY
+ // exclusions: DECIMAL, MAP, MULTISET, ROW, NULL, ANY
addDefaultDataType(String.class, DataTypes.STRING());
addDefaultDataType(Boolean.class, DataTypes.BOOLEAN());
addDefaultDataType(boolean.class, DataTypes.BOOLEAN());
@@ -66,6 +66,7 @@ public final class ClassDataTypeConverter {
addDefaultDataType(java.time.OffsetDateTime.class, DataTypes.TIMESTAMP_WITH_TIME_ZONE(9));
addDefaultDataType(java.time.Instant.class, DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9));
addDefaultDataType(java.time.Duration.class, DataTypes.INTERVAL(DataTypes.SECOND(9)));
+ addDefaultDataType(java.time.Period.class, DataTypes.INTERVAL(DataTypes.YEAR(4), DataTypes.MONTH()));
}
private static void addDefaultDataType(Class<?> clazz, DataType rootType) {
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
index bdf0dac..aa703a3 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
@@ -56,6 +56,18 @@ public class ClassDataTypeConverterTest {
{java.sql.Time.class, DataTypes.TIME(0).nullable().bridgedTo(java.sql.Time.class)},
+ {
+ java.time.Duration.class,
+ DataTypes.INTERVAL(DataTypes.SECOND(9))
+ .bridgedTo(java.time.Duration.class)
+ },
+
+ {
+ java.time.Period.class,
+ DataTypes.INTERVAL(DataTypes.YEAR(4), DataTypes.MONTH())
+ .bridgedTo(java.time.Period.class)
+ },
+
{BigDecimal.class, null},
{
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
new file mode 100644
index 0000000..bcfd20f
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
@@ -0,0 +1,815 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.HintFlag;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.DataTypeLookup;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.extraction.utils.DataTypeHintMock;
+import org.apache.flink.table.types.extraction.utils.DataTypeTemplate;
+import org.apache.flink.table.types.inference.utils.DataTypeLookupMock;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.StructuredType;
+import org.apache.flink.table.types.logical.TypeInformationRawType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+import static org.apache.flink.util.CoreMatchers.containsCause;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link DataTypeExtractor}.
+ */
+@RunWith(Parameterized.class)
+public class DataTypeExtractorTest {
+
+ @Parameters
+ public static List<TestSpec> testData() {
+ return Arrays.asList(
+ // simple extraction
+ TestSpec
+ .forType(Integer.class)
+ .expectDataType(DataTypes.INT()),
+
+ // extraction from hint conversion class
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public Class<?> bridgedTo() {
+ return Long.class;
+ }
+ },
+ Object.class)
+ .expectDataType(DataTypes.BIGINT()),
+
+ // missing precision/scale
+ TestSpec
+ .forType(BigDecimal.class)
+ .expectErrorMessage("Values of 'java.math.BigDecimal' need fixed precision and scale."),
+
+ // unsupported type exception
+ TestSpec
+ .forType(Object.class)
+ .expectErrorMessage(
+ "Could not extract a data type from 'class java.lang.Object'. " +
+ "Interpreting it as a structured type was also not successful."),
+
+ // explicit precision/scale through data type
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public String value() {
+ return "DECIMAL(5, 3)";
+ }
+ },
+ BigDecimal.class)
+ .expectDataType(DataTypes.DECIMAL(5, 3)),
+
+ // default precision/scale
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public int defaultDecimalPrecision() {
+ return 5;
+ }
+
+ @Override
+ public int defaultDecimalScale() {
+ return 3;
+ }
+ },
+ BigDecimal.class)
+ .expectDataType(DataTypes.DECIMAL(5, 3)),
+
+ // different bridging class via default conversion
+ TestSpec
+ .forType(java.sql.Timestamp.class)
+ .expectDataType(DataTypes.TIMESTAMP(9).bridgedTo(java.sql.Timestamp.class)),
+
+ // different bridging class via type itself
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public String value() {
+ return "TIMESTAMP(0)";
+ }
+ },
+ java.sql.Timestamp.class)
+ .expectDataType(DataTypes.TIMESTAMP(0).bridgedTo(java.sql.Timestamp.class)),
+
+ // default seconds for TIMESTAMP
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public int defaultSecondPrecision() {
+ return 6;
+ }
+ },
+ java.time.LocalDateTime.class)
+ .expectDataType(DataTypes.TIMESTAMP(6)),
+
+ // default second precision for INTERVAL
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public int defaultSecondPrecision() {
+ return 3;
+ }
+ },
+ java.time.Duration.class)
+ .expectDataType(DataTypes.INTERVAL(DataTypes.SECOND(3))),
+
+ // default year precision 2 for INTERVAL
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public int defaultYearPrecision() {
+ return 2;
+ }
+ },
+ java.time.Period.class)
+ .expectDataType(DataTypes.INTERVAL(DataTypes.YEAR(2), DataTypes.MONTH())),
+
+ // default year precision 0 for INTERVAL
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public int defaultYearPrecision() {
+ return 0;
+ }
+ },
+ java.time.Period.class)
+ .expectDataType(DataTypes.INTERVAL(DataTypes.MONTH())),
+
+ // RAW with default serializer
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public HintFlag allowRawGlobally() {
+ return HintFlag.TRUE;
+ }
+ },
+ Object[][].class)
+ .lookupExpects(Object.class)
+ .expectDataType(
+ DataTypes.ARRAY(
+ DataTypes.ARRAY(
+ DataTypes.RAW(new GenericTypeInfo<>(Object.class))))),
+
+ // RAW with custom serializer
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public String value() {
+ return "RAW";
+ }
+
+ @Override
+ public Class<? extends TypeSerializer<?>> rawSerializer() {
+ return IntSerializer.class;
+ }
+ },
+ Integer.class)
+ .expectDataType(DataTypes.RAW(Integer.class, new IntSerializer())),
+
+ // RAW with different conversion class
+ TestSpec
+ .forType(
+ new DataTypeHintMock() {
+ @Override
+ public String value() {
+ return "RAW";
+ }
+
+ @Override
+ public Class<?> bridgedTo() {
+ return Integer.class;
+ }
+ },
+ Object.class)
+ .lookupExpects(Integer.class)
+ .expectDataType(DataTypes.RAW(new GenericTypeInfo<>(Integer.class))),
+
+ // MAP type with type variable magic
+ TestSpec
+ .forGeneric(TableFunction.class, 0, TableFunctionWithMapLevel2.class)
+ .expectDataType(DataTypes.MAP(DataTypes.BIGINT(), DataTypes.BOOLEAN())),
+
+ // ARRAY type with type variable magic
+ TestSpec
+ .forGeneric(TableFunction.class, 0, TableFunctionWithGenericArray1.class)
+ .expectDataType(DataTypes.ARRAY(DataTypes.INT())),
+
+ // invalid type variable
+ TestSpec
+ .forGeneric(TableFunction.class, 0, TableFunctionWithHashMap.class)
+ .expectErrorMessage(
+ "Could not extract a data type from 'java.util.HashMap<java.lang.Integer, java.lang.String>'. " +
+ "Interpreting it as a structured type was also not successful."),
+
+ // simple structured type without RAW type
+ TestSpec
+ .forType(SimplePojo.class)
+ .expectDataType(getSimplePojoDataType(SimplePojo.class)),
+
+ // complex nested structured type annotation on top of type
+ TestSpec
+ .forType(ComplexPojo.class)
+ .lookupExpects(Object.class)
+ .expectDataType(getComplexPojoDataType(ComplexPojo.class, SimplePojo.class)),
+
+ // structured type with missing generics
+ TestSpec
+ .forType(SimplePojoWithGeneric.class)
+ .expectErrorMessage(
+ "Unresolved type variable 'S'. A data type cannot be extracted from a type variable. " +
+ "The original content might have been erased due to Java type erasure."),
+
+ // structured type with annotation on top of field
+ TestSpec
+ .forGeneric(TableFunction.class, 0, TableFunctionWithGenericPojo.class)
+ .lookupExpects(Object.class)
+ .expectDataType(getComplexPojoDataType(ComplexPojoWithGeneric.class, SimplePojoWithGeneric.class)),
+
+ // extraction with generic interfaces
+ TestSpec
+ .forGeneric(BaseInterface.class, 0, ConcreteClass.class)
+ .expectDataType(DataTypes.STRING()),
+
+ // structured type with hierarchy
+ TestSpec
+ .forType(SimplePojoWithGenericHierarchy.class)
+ .expectDataType(getSimplePojoDataType(SimplePojoWithGenericHierarchy.class)),
+
+ // structured type with different getter and setter flavors
+ TestSpec
+ .forType(ComplexPojoWithGettersAndSetters.class)
+ .lookupExpects(Object.class)
+ .expectDataType(getComplexPojoDataType(ComplexPojoWithGettersAndSetters.class, SimplePojo.class)),
+
+ // structure type with missing setter
+ TestSpec
+ .forType(SimplePojoWithMissingSetter.class)
+ .expectErrorMessage(
+ "Field 'stringField' of class '" + SimplePojoWithMissingSetter.class.getName() + "' is " +
+ "mutable but is neither publicly accessible nor does it have a corresponding setter method."),
+
+ // structure type with missing getter
+ TestSpec
+ .forType(SimplePojoWithMissingGetter.class)
+ .expectErrorMessage(
+ "Field 'stringField' of class '" + SimplePojoWithMissingGetter.class.getName() + "' is " +
+ "neither publicly accessible nor does it have a corresponding getter method."),
+
+ // structured type with assigning constructor
+ TestSpec
+ .forType(SimplePojoWithAssigningConstructor.class)
+ .expectDataType(getSimplePojoDataType(SimplePojoWithAssigningConstructor.class)),
+
+ // assigning constructor defines field order
+ TestSpec
+ .forType(PojoWithCustomFieldOrder.class)
+ .expectDataType(getPojoWithCustomOrderDataType(PojoWithCustomFieldOrder.class)),
+
+ // Flink tuples
+ TestSpec
+ .forGeneric(TableFunction.class, 0, TableFunctionWithTuples.class)
+ .expectDataType(getOuterTupleDataType()),
+
+ // many annotations that partially override each other
+ TestSpec
+ .forType(SimplePojoWithManyAnnotations.class)
+ .expectDataType(getSimplePojoDataType(SimplePojoWithManyAnnotations.class)),
+
+ // many annotations that partially override each other
+ TestSpec
+ .forType(ComplexPojoWithManyAnnotations.class)
+ .lookupExpects(Object.class)
+ .expectDataType(getComplexPojoDataType(ComplexPojoWithManyAnnotations.class, SimplePojo.class))
+ );
+ }
+
+ @Parameter
+ public TestSpec testSpec;
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void testExtraction() {
+ if (testSpec.expectedErrorMessage != null) {
+ thrown.expect(ValidationException.class);
+ thrown.expectCause(containsCause(new ValidationException(testSpec.expectedErrorMessage)));
+ }
+ runExtraction(testSpec);
+ }
+
+ static void runExtraction(TestSpec testSpec) {
+ final DataType dataType = testSpec.extractor.apply(testSpec.lookup);
+ if (testSpec.expectedDataType != null) {
+ assertThat(dataType, equalTo(testSpec.expectedDataType));
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Test utilities
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Test specification shared with the Scala tests.
+ */
+ static class TestSpec {
+
+ private final Function<DataTypeLookup, DataType> extractor;
+
+ private DataTypeLookupMock lookup;
+
+ private @Nullable DataType expectedDataType;
+
+ private @Nullable String expectedErrorMessage;
+
+ private TestSpec(Function<DataTypeLookup, DataType> extractor) {
+ this.lookup = new DataTypeLookupMock();
+ this.extractor = extractor;
+ }
+
+ static TestSpec forType(Type type) {
+ return new TestSpec((lookup) -> DataTypeExtractor.extractFromType(lookup, type));
+ }
+
+ static TestSpec forType(DataTypeHint hint, Type type) {
+ return new TestSpec((lookup) ->
+ DataTypeExtractor.extractFromType(lookup, DataTypeTemplate.fromAnnotation(lookup, hint), type));
+ }
+
+ static TestSpec forGeneric(Class<?> baseClass, int genericPos, Type contextType) {
+ return new TestSpec((lookup) ->
+ DataTypeExtractor.extractFromGeneric(lookup, baseClass, genericPos, contextType));
+ }
+
+ static TestSpec forMethodParameter(Class<?> clazz, int paramPos) {
+ final Method method = clazz.getDeclaredMethods()[0];
+ return new TestSpec((lookup) ->
+ DataTypeExtractor.extractFromMethodParameter(lookup, clazz, method, paramPos));
+ }
+
+ static TestSpec forMethodOutput(Class<?> clazz) {
+ final Method method = clazz.getDeclaredMethods()[0];
+ return new TestSpec((lookup) ->
+ DataTypeExtractor.extractFromMethodOutput(lookup, clazz, method));
+ }
+
+ TestSpec lookupExpects(Class<?> lookupClass) {
+ lookup.dataType = Optional.of(DataTypes.RAW(new GenericTypeInfo<>(lookupClass)));
+ lookup.expectedClass = Optional.of(lookupClass);
+ return this;
+ }
+
+ TestSpec expectDataType(DataType expectedDataType) {
+ this.expectedDataType = expectedDataType;
+ return this;
+ }
+
+ TestSpec expectErrorMessage(String expectedErrorMessage) {
+ this.expectedErrorMessage = expectedErrorMessage;
+ return this;
+ }
+ }
+
+ /**
+ * Testing data type shared with the Scala tests.
+ */
+ static DataType getSimplePojoDataType(Class<?> simplePojoClass) {
+ final StructuredType.Builder builder = StructuredType.newBuilder(simplePojoClass);
+ builder.attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute("intField", new IntType(true)),
+ new StructuredType.StructuredAttribute("primitiveBooleanField", new BooleanType(false)),
+ new StructuredType.StructuredAttribute("primitiveIntField", new IntType(false)),
+ new StructuredType.StructuredAttribute("stringField", new VarCharType(VarCharType.MAX_LENGTH))));
+ builder.setFinal(true);
+ builder.setInstantiable(true);
+ final StructuredType structuredType = builder.build();
+
+ final Map<String, DataType> fields = new HashMap<>();
+ fields.put("intField", DataTypes.INT());
+ fields.put("primitiveBooleanField", DataTypes.BOOLEAN().notNull().bridgedTo(boolean.class));
+ fields.put("primitiveIntField", DataTypes.INT().notNull().bridgedTo(int.class));
+ fields.put("stringField", DataTypes.STRING());
+
+ return new FieldsDataType(structuredType, simplePojoClass, fields);
+ }
+
+ /**
+ * Testing data type shared with the Scala tests.
+ */
+ static DataType getComplexPojoDataType(Class<?> complexPojoClass, Class<?> simplePojoClass) {
+ final StructuredType.Builder builder = StructuredType.newBuilder(complexPojoClass);
+ builder.attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute(
+ "mapField",
+ new MapType(new VarCharType(VarCharType.MAX_LENGTH), new IntType())),
+ new StructuredType.StructuredAttribute(
+ "simplePojoField",
+ getSimplePojoDataType(simplePojoClass).getLogicalType()),
+ new StructuredType.StructuredAttribute(
+ "someObject",
+ new TypeInformationRawType<>(new GenericTypeInfo<>(Object.class)))));
+ builder.setFinal(true);
+ builder.setInstantiable(true);
+ final StructuredType structuredType = builder.build();
+
+ final Map<String, DataType> fields = new HashMap<>();
+ fields.put("mapField", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()));
+ fields.put("simplePojoField", getSimplePojoDataType(simplePojoClass));
+ fields.put("someObject", DataTypes.RAW(new GenericTypeInfo<>(Object.class)));
+
+ return new FieldsDataType(structuredType, complexPojoClass, fields);
+ }
+
+ /**
+ * Testing data type shared with the Scala tests.
+ */
+ public static DataType getPojoWithCustomOrderDataType(Class<?> pojoClass) {
+ final StructuredType.Builder builder = StructuredType.newBuilder(pojoClass);
+ builder.attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute(
+ "z",
+ new BigIntType()),
+ new StructuredType.StructuredAttribute(
+ "y",
+ new BooleanType()),
+ new StructuredType.StructuredAttribute(
+ "x",
+ new IntType())));
+ builder.setFinal(true);
+ builder.setInstantiable(true);
+ final StructuredType structuredType = builder.build();
+
+ final Map<String, DataType> fields = new HashMap<>();
+ fields.put("z", DataTypes.BIGINT());
+ fields.put("y", DataTypes.BOOLEAN());
+ fields.put("x", DataTypes.INT());
+
+ return new FieldsDataType(structuredType, pojoClass, fields);
+ }
+
+ private static DataType getOuterTupleDataType() {
+ final StructuredType.Builder builder = StructuredType.newBuilder(Tuple2.class);
+ builder.attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute(
+ "f0",
+ new IntType()),
+ new StructuredType.StructuredAttribute(
+ "f1",
+ getInnerTupleDataType().getLogicalType())));
+ builder.setFinal(true);
+ builder.setInstantiable(true);
+ final StructuredType structuredType = builder.build();
+
+ final Map<String, DataType> fields = new HashMap<>();
+ fields.put("f0", DataTypes.INT());
+ fields.put("f1", getInnerTupleDataType());
+
+ return new FieldsDataType(structuredType, Tuple2.class, fields);
+ }
+
+ private static DataType getInnerTupleDataType() {
+ final StructuredType.Builder builder = StructuredType.newBuilder(Tuple2.class);
+ builder.attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute(
+ "f0",
+ new VarCharType(VarCharType.MAX_LENGTH)),
+ new StructuredType.StructuredAttribute(
+ "f1",
+ new BooleanType())));
+ builder.setFinal(true);
+ builder.setInstantiable(true);
+ final StructuredType structuredType = builder.build();
+
+ final Map<String, DataType> fields = new HashMap<>();
+ fields.put("f0", DataTypes.STRING());
+ fields.put("f1", DataTypes.BOOLEAN());
+
+ return new FieldsDataType(structuredType, Tuple2.class, fields);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Test classes for extraction
+ // --------------------------------------------------------------------------------------------
+
+ private static class TableFunctionWithMapLevel0<K, V> extends TableFunction<Map<K, V>> {
+
+ }
+
+ private static class TableFunctionWithMapLevel1<V> extends TableFunctionWithMapLevel0<Long, V> {
+
+ }
+
+ private static class TableFunctionWithMapLevel2 extends TableFunctionWithMapLevel1<Boolean> {
+
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static class TableFunctionWithGenericArray0<T> extends TableFunction<T[]> {
+
+ }
+
+ private static class TableFunctionWithGenericArray1 extends TableFunctionWithGenericArray0<Integer> {
+
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static class TableFunctionWithHashMap extends TableFunction<HashMap<Integer, String>> {
+
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Complex POJO with raw types.
+ */
+ @SuppressWarnings("unused")
+ @DataTypeHint(allowRawGlobally = HintFlag.TRUE)
+ public static class ComplexPojo {
+ public Map<String, Integer> mapField;
+ public SimplePojo simplePojoField;
+ public Object someObject;
+ }
+
+ /**
+ * Simple POJO with no RAW types.
+ */
+ @SuppressWarnings("unused")
+ public static class SimplePojo {
+ public Integer intField;
+ public boolean primitiveBooleanField;
+ public int primitiveIntField;
+ public String stringField;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static class TableFunctionWithGenericPojo extends TableFunction<ComplexPojoWithGeneric<String, Integer>> {
+
+ }
+
+ /**
+ * In the end this should be the same data type as {@link ComplexPojo}.
+ */
+ @SuppressWarnings("unused")
+ public static class ComplexPojoWithGeneric<S, I> {
+ public Map<S, I> mapField;
+ public SimplePojoWithGeneric<S> simplePojoField;
+ @DataTypeHint(allowRawGlobally = HintFlag.TRUE)
+ public Object someObject;
+ }
+
+ /**
+ * In the end this should be the same data type as {@link SimplePojo}.
+ */
+ @SuppressWarnings("unused")
+ public static class SimplePojoWithGeneric<S> {
+ public Integer intField;
+ public boolean primitiveBooleanField;
+ public int primitiveIntField;
+ public S stringField;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private interface BaseInterface<T> {
+ // no implementation
+ }
+
+ private abstract static class GenericClass<T> implements Serializable, BaseInterface<T> {
+ // no implementation
+ }
+
+ private static class ConcreteClass extends GenericClass<String> {
+ // no implementation
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Hierarchy of structured type.
+ */
+ public static class SimplePojoWithGenericHierarchy extends SimplePojoWithGeneric<String> {
+ // no implementation
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Private member variables.
+ */
+ public static class ComplexPojoWithGettersAndSetters {
+ private Map<String, Integer> mapField;
+ private SimplePojo simplePojoField;
+ private @DataTypeHint("RAW") Object someObject;
+
+ // Java-like
+ public Map<String, Integer> getMapField() {
+ return mapField;
+ }
+
+ public void setMapField(Map<String, Integer> mapField) {
+ this.mapField = mapField;
+ }
+
+ public SimplePojo getSimplePojoField() {
+ return simplePojoField;
+ }
+
+ public void setSimplePojoField(SimplePojo simplePojoField) {
+ this.simplePojoField = simplePojoField;
+ }
+
+ // Scala-like
+ public Object someObject() {
+ return someObject;
+ }
+
+ public void someObject(Object someObject) {
+ this.someObject = someObject;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * {@code setStringField()} is missing.
+ */
+ @SuppressWarnings("unused")
+ public static class SimplePojoWithMissingSetter {
+ public Integer intField;
+ public boolean primitiveBooleanField;
+ public int primitiveIntField;
+ private String stringField;
+
+ public String getStringField() {
+ return stringField;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * {@code getStringField()} is missing.
+ */
+ @SuppressWarnings({"FieldCanBeLocal", "unused"})
+ public static class SimplePojoWithMissingGetter {
+ public Integer intField;
+ public boolean primitiveBooleanField;
+ public int primitiveIntField;
+ private String stringField;
+
+ public void setStringField(String stringField) {
+ this.stringField = stringField;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Immutable type.
+ */
+ public static class SimplePojoWithAssigningConstructor {
+ public final Integer intField;
+ public final boolean primitiveBooleanField;
+ public final int primitiveIntField;
+ public final String stringField;
+
+ public SimplePojoWithAssigningConstructor(
+ Integer intField,
+ boolean primitiveBooleanField,
+ int primitiveIntField,
+ String stringField) {
+ this.intField = intField;
+ this.primitiveBooleanField = primitiveBooleanField;
+ this.primitiveIntField = primitiveIntField;
+ this.stringField = stringField;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static class TableFunctionWithTuples extends TableFunction<Tuple2<Integer, Tuple2<String, Boolean>>> {
+
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Constructor reorders fields.
+ */
+ public static class PojoWithCustomFieldOrder {
+
+ public Integer x;
+ public Boolean y;
+ public Long z;
+
+ public PojoWithCustomFieldOrder(long z, boolean y, int x) {
+ this.z = z;
+ this.y = y;
+ this.x = x;
+ }
+
+ @SuppressWarnings("unused")
+ public PojoWithCustomFieldOrder(long z, boolean y, int x, int additional) {
+ this(z, y, x);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * In the end this should be the same data type as {@link SimplePojo}.
+ */
+ @SuppressWarnings("unused")
+ @DataTypeHint(forceRawPattern = {"java.lang."})
+ public static class SimplePojoWithManyAnnotations {
+ public @DataTypeHint("INT") Integer intField;
+ public @DataTypeHint(bridgedTo = boolean.class) Object primitiveBooleanField;
+ public @DataTypeHint(value = "INT NOT NULL", bridgedTo = int.class) Object primitiveIntField;
+ @DataTypeHint(forceRawPattern = {})
+ public String stringField;
+ }
+
+ /**
+ * In the end this should be the same data type as {@link ComplexPojo}.
+ */
+ @SuppressWarnings("unused")
+ @DataTypeHint(allowRawPattern = {"java.lang"})
+ public static class ComplexPojoWithManyAnnotations {
+ public @DataTypeHint("MAP<STRING, INT>") Object mapField;
+ public SimplePojo simplePojoField;
+ public Object someObject;
+ }
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/utils/DataTypeHintMock.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/utils/DataTypeHintMock.java
new file mode 100644
index 0000000..76e5099
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/utils/DataTypeHintMock.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction.utils;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ExtractionVersion;
+import org.apache.flink.table.annotation.HintFlag;
+import org.apache.flink.table.annotation.InputGroup;
+
+import java.lang.annotation.Annotation;
+
+/**
+ * Mock for {@link DataTypeHint}. Methods can be overridden if necessary.
+ */
+@SuppressWarnings("ClassExplicitlyAnnotation")
+public class DataTypeHintMock implements DataTypeHint {
+
+ private static final DataTypeHint DEFAULT_ANNOTATION = getDefaultAnnotation();
+
+ @Override
+ public String value() {
+ return DEFAULT_ANNOTATION.value();
+ }
+
+ @Override
+ public Class<?> bridgedTo() {
+ return DEFAULT_ANNOTATION.bridgedTo();
+ }
+
+ @Override
+ public Class<? extends TypeSerializer<?>> rawSerializer() {
+ return DEFAULT_ANNOTATION.rawSerializer();
+ }
+
+ @Override
+ public InputGroup inputGroup() {
+ return DEFAULT_ANNOTATION.inputGroup();
+ }
+
+ @Override
+ public ExtractionVersion version() {
+ return DEFAULT_ANNOTATION.version();
+ }
+
+ @Override
+ public HintFlag allowRawGlobally() {
+ return DEFAULT_ANNOTATION.allowRawGlobally();
+ }
+
+ @Override
+ public String[] allowRawPattern() {
+ return DEFAULT_ANNOTATION.allowRawPattern();
+ }
+
+ @Override
+ public String[] forceRawPattern() {
+ return DEFAULT_ANNOTATION.forceRawPattern();
+ }
+
+ @Override
+ public int defaultDecimalPrecision() {
+ return DEFAULT_ANNOTATION.defaultDecimalPrecision();
+ }
+
+ @Override
+ public int defaultDecimalScale() {
+ return DEFAULT_ANNOTATION.defaultDecimalScale();
+ }
+
+ @Override
+ public int defaultYearPrecision() {
+ return DEFAULT_ANNOTATION.defaultYearPrecision();
+ }
+
+ @Override
+ public int defaultSecondPrecision() {
+ return DEFAULT_ANNOTATION.defaultSecondPrecision();
+ }
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return DataTypeHint.class;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @DataTypeHint
+ private static class DefaultAnnotationHelper {
+ // no implementation
+ }
+
+ private static DataTypeHint getDefaultAnnotation() {
+ return DefaultAnnotationHelper.class.getAnnotation(DataTypeHint.class);
+ }
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/DataTypeLookupMock.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/DataTypeLookupMock.java
new file mode 100644
index 0000000..1bd5997
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/DataTypeLookupMock.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.utils;
+
+import org.apache.flink.table.catalog.DataTypeLookup;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeParser;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.Optional;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * {@link DataTypeLookup} mock for testing purposes.
+ */
+public class DataTypeLookupMock implements DataTypeLookup {
+
+ public Optional<DataType> dataType = Optional.empty();
+
+ public Optional<Class<?>> expectedClass = Optional.empty();
+
+ @Override
+ public Optional<DataType> lookupDataType(String name) {
+ return Optional.of(TypeConversions.fromLogicalToDataType(LogicalTypeParser.parse(name)));
+ }
+
+ @Override
+ public Optional<DataType> lookupDataType(UnresolvedIdentifier identifier) {
+ return dataType;
+ }
+
+ @Override
+ public DataType resolveRawDataType(Class<?> clazz) {
+ expectedClass.ifPresent(expected -> assertEquals(expected, clazz));
+ return dataType.orElseThrow(IllegalStateException::new);
+ }
+}