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/06/05 07:51:22 UTC
[flink] 01/02: [FLINK-12254][table] Update TableSource and related
interfaces to new type system
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
commit cedd9803a9d7e3312e49fa46025ce9ac985bf2a0
Author: Timo Walther <tw...@apache.org>
AuthorDate: Fri May 31 17:13:42 2019 +0200
[FLINK-12254][table] Update TableSource and related interfaces to new type system
---
.../flink/table/sources/DefinedFieldMapping.java | 18 +++++-----
.../NestedFieldsProjectableTableSource.java | 10 +++---
.../table/sources/ProjectableTableSource.java | 8 ++---
.../apache/flink/table/sources/TableSource.java | 36 +++++++++++++++----
.../table/plan/nodes/common/CommonLookupJoin.scala | 27 ++++++++-------
.../physical/batch/BatchExecTableSourceScan.scala | 37 +++++++++++---------
.../stream/StreamExecTableSourceScan.scala | 40 ++++++++++++----------
.../flink/table/sources/TableSourceUtil.scala | 6 ++--
.../plan/nodes/dataset/BatchTableSourceScan.scala | 10 ++++--
.../nodes/datastream/StreamTableSourceScan.scala | 10 ++++--
.../flink/table/sources/TableSourceUtil.scala | 10 +++---
.../table/catalog/CatalogStructureBuilder.java | 12 ++++---
12 files changed, 134 insertions(+), 90 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedFieldMapping.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedFieldMapping.java
index dd1a0ca..daa7924 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedFieldMapping.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedFieldMapping.java
@@ -19,8 +19,8 @@
package org.apache.flink.table.sources;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
@@ -28,11 +28,11 @@ import java.util.Map;
/**
* The {@link DefinedFieldMapping} interface provides a mapping for the fields of the table schema
- * ({@link TableSource#getTableSchema} to fields of the physical returned type
- * {@link TableSource#getReturnType} of a {@link TableSource}.
+ * ({@link TableSource#getTableSchema} to fields of the physical produced data type
+ * {@link TableSource#getProducedDataType()} of a {@link TableSource}.
*
* <p>If a {@link TableSource} does not implement the {@link DefinedFieldMapping} interface, the fields of
- * its {@link TableSchema} are mapped to the fields of its return type {@link TypeInformation} by name.
+ * its {@link TableSchema} are mapped to the fields of its produced {@link DataType} by name.
*
* <p>If the fields cannot or should not be implicitly mapped by name, an explicit mapping can be
* provided by implementing this interface.
@@ -44,17 +44,17 @@ public interface DefinedFieldMapping {
/**
* Returns the mapping for the fields of the {@link TableSource}'s {@link TableSchema} to the fields of
- * its return type {@link TypeInformation}.
+ * its produced {@link DataType}.
*
* <p>The mapping is done based on field names, e.g., a mapping "name" -> "f1" maps the schema field
- * "name" to the field "f1" of the return type, for example in this case the second field of a
+ * "name" to the field "f1" of the produced data type, for example in this case the second field of a
* {@link org.apache.flink.api.java.tuple.Tuple}.
*
- * <p>The returned mapping must map all fields (except proctime and rowtime fields) to the return
+ * <p>The returned mapping must map all fields (except proctime and rowtime fields) to the produced data
* type. It can also provide a mapping for fields which are not in the {@link TableSchema} to make
- * fields in the physical {@link TypeInformation} accessible for a {@code TimestampExtractor}.
+ * fields in the physical {@link DataType} accessible for a {@code TimestampExtractor}.
*
- * @return A mapping from {@link TableSchema} fields to {@link TypeInformation} fields or
+ * @return A mapping from {@link TableSchema} fields to {@link DataType} fields or
* null if no mapping is necessary.
*/
@Nullable
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/NestedFieldsProjectableTableSource.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/NestedFieldsProjectableTableSource.java
index e161c78..49fb22b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/NestedFieldsProjectableTableSource.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/NestedFieldsProjectableTableSource.java
@@ -34,13 +34,13 @@ public interface NestedFieldsProjectableTableSource<T> {
/**
* Creates a copy of the {@link TableSource} that projects its output to the given field indexes.
- * The field indexes relate to the physical return type ({@link TableSource#getReturnType()}) and not
- * to the table schema ({@link TableSource#getTableSchema()} of the {@link TableSource}.
+ * The field indexes relate to the physical produced data type ({@link TableSource#getProducedDataType()})
+ * and not to the table schema ({@link TableSource#getTableSchema()} of the {@link TableSource}.
*
* <p>The table schema ({@link TableSource#getTableSchema()} of the {@link TableSource} copy must not be
- * modified by this method, but only the return type ({@link TableSource#getReturnType()}) and the
- * produced {@code DataSet} ({@code BatchTableSource.getDataSet(}) or {@code DataStream}
- * ({@code StreamTableSource.getDataStream()}). The return type may only be changed by
+ * modified by this method, but only the produced data type ({@link TableSource#getProducedDataType()})
+ * and the produced {@code DataSet} ({@code BatchTableSource.getDataSet(}) or {@code DataStream}
+ * ({@code StreamTableSource.getDataStream()}). The produced data type may only be changed by
* removing or reordering first level fields. The type of the first level fields must not be
* changed.
*
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/ProjectableTableSource.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/ProjectableTableSource.java
index 1c4b3c1..e79951e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/ProjectableTableSource.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/ProjectableTableSource.java
@@ -34,12 +34,12 @@ public interface ProjectableTableSource<T> {
/**
* Creates a copy of the {@link TableSource} that projects its output to the given field indexes.
- * The field indexes relate to the physical return type ({@link TableSource#getReturnType}) and not
- * to the table schema ({@link TableSource#getTableSchema} of the {@link TableSource}.
+ * The field indexes relate to the physical poduced data type ({@link TableSource#getProducedDataType()})
+ * and not to the table schema ({@link TableSource#getTableSchema} of the {@link TableSource}.
*
* <p>The table schema ({@link TableSource#getTableSchema} of the {@link TableSource} copy must not be
- * modified by this method, but only the return type ({@link TableSource#getReturnType}) and the
- * produced {@code DataSet} ({@code BatchTableSource#getDataSet(}) or {@code DataStream}
+ * modified by this method, but only the produced data type ({@link TableSource#getProducedDataType()})
+ * and the produced {@code DataSet} ({@code BatchTableSource#getDataSet(}) or {@code DataStream}
* ({@code StreamTableSource#getDataStream}).
*
* <p>If the {@link TableSource} implements the {@link DefinedFieldMapping} interface, it might
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/TableSource.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/TableSource.java
index 05bf92c..bf379f6 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/TableSource.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/TableSource.java
@@ -20,18 +20,23 @@ package org.apache.flink.table.sources;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.utils.TableConnectorUtils;
+import static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType;
+
/**
* Defines an external table with the schema that is provided by {@link TableSource#getTableSchema}.
*
* <p>The data of a {@link TableSource} is produced as a {@code DataSet} in case of a {@code BatchTableSource}
* or as a {@code DataStream} in case of a {@code StreamTableSource}. The type of ths produced
- * {@code DataSet} or {@code DataStream} is specified by the {@link TableSource#getReturnType} method.
+ * {@code DataSet} or {@code DataStream} is specified by the {@link TableSource#getProducedDataType()} method.
*
* <p>By default, the fields of the {@link TableSchema} are implicitly mapped by name to the fields of
- * the return type {@link TypeInformation}. An explicit mapping can be defined by implementing the
+ * the produced {@link DataType}. An explicit mapping can be defined by implementing the
* {@link DefinedFieldMapping} interface.
*
* @param <T> The return type of the {@link TableSource}.
@@ -40,12 +45,31 @@ import org.apache.flink.table.utils.TableConnectorUtils;
public interface TableSource<T> {
/**
- * Returns the {@link TypeInformation} for the return type of the {@link TableSource}.
- * The fields of the return type are mapped to the table schema based on their name.
+ * Returns the {@link DataType} for the produced data of the {@link TableSource}.
+ * The fields of the data type are mapped to the table schema based on their name.
*
- * @return The type of the returned {@code DataSet} or {@code DataStream}.
+ * @return The data type of the returned {@code DataSet} or {@code DataStream}.
+ */
+ default DataType getProducedDataType() {
+ final TypeInformation<T> legacyType = getReturnType();
+ if (legacyType == null) {
+ throw new TableException("Table source does not implement a produced data type.");
+ }
+ return fromLegacyInfoToDataType(getReturnType());
+ }
+
+ /**
+ * @deprecated This method will be removed in future versions as it uses the old type system. It
+ * is recommended to use {@link #getProducedDataType()} instead which uses the new type
+ * system based on {@link DataTypes}. Please make sure to use either the old or the new type
+ * system consistently to avoid unintended behavior. See the website documentation
+ * for more information.
*/
- TypeInformation<T> getReturnType();
+ @Deprecated
+ @SuppressWarnings("unchecked")
+ default TypeInformation<T> getReturnType() {
+ return null;
+ }
/**
* Returns the schema of the produced table.
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonLookupJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonLookupJoin.scala
index 6d2e40d..695e116 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonLookupJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonLookupJoin.scala
@@ -17,6 +17,8 @@
*/
package org.apache.flink.table.plan.nodes.common
+import java.util.Collections
+
import com.google.common.primitives.Primitives
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
@@ -45,17 +47,15 @@ import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getParamClassesConsiderVarArgs, getUserDefinedMethod, signatureToString, signaturesToString}
import org.apache.flink.table.functions.{AsyncTableFunction, TableFunction, UserDefinedFunction}
import org.apache.flink.table.plan.nodes.FlinkRelNode
-import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
-import org.apache.flink.table.plan.util.{JoinTypeUtil, RelExplainUtil}
import org.apache.flink.table.plan.util.LookupJoinUtil._
-import org.apache.flink.table.runtime.join.lookup.{AsyncLookupJoinRunner, LookupJoinRunner, AsyncLookupJoinWithCalcRunner, LookupJoinWithCalcRunner}
+import org.apache.flink.table.plan.util.{JoinTypeUtil, RelExplainUtil}
+import org.apache.flink.table.runtime.join.lookup.{AsyncLookupJoinRunner, AsyncLookupJoinWithCalcRunner, LookupJoinRunner, LookupJoinWithCalcRunner}
import org.apache.flink.table.sources.TableIndex.IndexType
import org.apache.flink.table.sources.{LookupConfig, LookupableTableSource, TableIndex, TableSource}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.BaseRowTypeInfo
import org.apache.flink.types.Row
-import java.util.Collections
-
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -157,6 +157,9 @@ abstract class CommonLookupJoin(
val resultRowType = FlinkTypeFactory.toInternalRowType(getRowType)
val tableSchema = tableSource.getTableSchema
+ val producedDataType = tableSource.getProducedDataType
+ val producedTypeInfo = fromDataTypeToLegacyInfo(producedDataType)
+
// validate whether the node is valid and supported.
validate(
tableSource,
@@ -202,7 +205,7 @@ abstract class CommonLookupJoin(
0)
checkUdtfReturnType(
tableSource.explainSource(),
- tableSource.getReturnType,
+ producedTypeInfo,
udtfResultType,
extractedResultTypeInfo)
val parameters = Array(new GenericType(classOf[ResultFuture[_]])) ++ lookupFieldTypesInOrder
@@ -216,7 +219,7 @@ abstract class CommonLookupJoin(
relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory],
inputRowType,
resultRowType,
- tableSource.getReturnType,
+ producedTypeInfo,
lookupFieldsInOrder,
allLookupKeys,
asyncLookupFunction)
@@ -240,7 +243,7 @@ abstract class CommonLookupJoin(
generatedFetcher,
generatedCalc,
generatedResultFuture,
- tableSource.getReturnType,
+ producedTypeInfo,
rightRowType.toTypeInfo,
leftOuterJoin,
lookupConfig.getAsyncBufferCapacity)
@@ -256,7 +259,7 @@ abstract class CommonLookupJoin(
new AsyncLookupJoinRunner(
generatedFetcher,
generatedResultFuture,
- tableSource.getReturnType,
+ producedTypeInfo,
rightRowType.toTypeInfo,
leftOuterJoin,
asyncBufferCapacity)
@@ -277,7 +280,7 @@ abstract class CommonLookupJoin(
0)
checkUdtfReturnType(
tableSource.explainSource(),
- tableSource.getReturnType,
+ producedTypeInfo,
udtfResultType,
extractedResultTypeInfo)
checkEvalMethodSignature(
@@ -290,7 +293,7 @@ abstract class CommonLookupJoin(
relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory],
inputRowType,
resultRowType,
- tableSource.getReturnType,
+ producedTypeInfo,
lookupFieldsInOrder,
allLookupKeys,
lookupFunction,
@@ -655,7 +658,7 @@ abstract class CommonLookupJoin(
"but was " + joinType.toString + " JOIN")
}
- val tableReturnType = tableSource.getReturnType
+ val tableReturnType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
if (!tableReturnType.isInstanceOf[BaseRowTypeInfo] &&
!tableReturnType.isInstanceOf[RowTypeInfo]) {
throw new TableException(
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecTableSourceScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecTableSourceScan.scala
index ee39511..ade92db 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecTableSourceScan.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecTableSourceScan.scala
@@ -18,25 +18,25 @@
package org.apache.flink.table.plan.nodes.physical.batch
+import java.util
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rex.RexNode
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.runtime.operators.DamBehavior
import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.table.api.{BatchTableEnvironment, TableException, Types}
+import org.apache.flink.table.codegen.CodeGeneratorContext
import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.nodes.physical.PhysicalTableSourceScan
import org.apache.flink.table.plan.schema.FlinkRelOptTable
-import org.apache.flink.table.sources.{BatchTableSource, TableSourceUtil}
-import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
-import org.apache.flink.table.codegen.CodeGeneratorContext
import org.apache.flink.table.plan.util.ScanUtil
-
-import org.apache.calcite.plan._
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.metadata.RelMetadataQuery
-import org.apache.calcite.rex.RexNode
-
-import java.util
+import org.apache.flink.table.sources.{BatchTableSource, TableSourceUtil}
+import org.apache.flink.table.types.utils.TypeConversions
+import org.apache.flink.table.types.utils.TypeConversions.{fromDataTypeToLegacyInfo, fromLegacyInfoToDataType}
import scala.collection.JavaConversions._
@@ -89,13 +89,16 @@ class BatchExecTableSourceScan(
isStreamTable = false,
None)
+ val inputDataType = fromLegacyInfoToDataType(inputTransform.getOutputType)
+ val producedDataType = tableSource.getProducedDataType
+ val producedTypeInfo = fromDataTypeToLegacyInfo(producedDataType)
+
// check that declared and actual type of table source DataStream are identical
- if (createInternalTypeFromTypeInfo(inputTransform.getOutputType) !=
- createInternalTypeFromTypeInfo(tableSource.getReturnType)) {
+ if (inputDataType != producedDataType) {
throw new TableException(s"TableSource of type ${tableSource.getClass.getCanonicalName} " +
- s"returned a DataSet of type ${inputTransform.getOutputType} that does not match with " +
- s"the type ${tableSource.getReturnType} declared by the TableSource.getReturnType() " +
- s"method. Please validate the implementation of the TableSource.")
+ s"returned a DataStream of data type $producedDataType that does not match with the " +
+ s"data type $producedDataType declared by the TableSource.getProducedDataType() method. " +
+ s"Please validate the implementation of the TableSource.")
}
// get expression to extract rowtime attribute
@@ -111,7 +114,7 @@ class BatchExecTableSourceScan(
CodeGeneratorContext(config),
inputTransform.asInstanceOf[StreamTransformation[Any]],
fieldIndexes,
- tableSource.getReturnType,
+ producedTypeInfo,
getRowType,
getTable.getQualifiedName,
config,
@@ -129,7 +132,7 @@ class BatchExecTableSourceScan(
None)
ScanUtil.hasTimeAttributeField(fieldIndexes) ||
ScanUtil.needsConversion(
- tableSource.getReturnType,
+ fromDataTypeToLegacyInfo(tableSource.getProducedDataType),
TypeExtractor.createTypeInfo(
tableSource, classOf[BatchTableSource[_]], tableSource.getClass, 0)
.getTypeClass.asInstanceOf[Class[_]])
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTableSourceScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
index 096604b..0579f9e 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
@@ -18,30 +18,29 @@
package org.apache.flink.table.plan.nodes.physical.stream
+import java.util
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rex.RexNode
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.table.api.{StreamTableEnvironment, TableException, Types}
+import org.apache.flink.table.codegen.CodeGeneratorContext
+import org.apache.flink.table.codegen.OperatorCodeGenerator._
import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
import org.apache.flink.table.plan.nodes.physical.PhysicalTableSourceScan
import org.apache.flink.table.plan.schema.FlinkRelOptTable
-import org.apache.flink.table.sources.wmstrategies.{PeriodicWatermarkAssigner, PreserveWatermarks, PunctuatedWatermarkAssigner}
-import org.apache.flink.table.sources.{RowtimeAttributeDescriptor, StreamTableSource, TableSourceUtil}
-import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
-import org.apache.flink.table.codegen.CodeGeneratorContext
-import org.apache.flink.table.codegen.OperatorCodeGenerator._
import org.apache.flink.table.plan.util.ScanUtil
import org.apache.flink.table.runtime.AbstractProcessStreamOperator
-
-import org.apache.calcite.plan._
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.metadata.RelMetadataQuery
-import org.apache.calcite.rex.RexNode
-
-import java.util
+import org.apache.flink.table.sources.wmstrategies.{PeriodicWatermarkAssigner, PreserveWatermarks, PunctuatedWatermarkAssigner}
+import org.apache.flink.table.sources.{RowtimeAttributeDescriptor, StreamTableSource, TableSourceUtil}
+import org.apache.flink.table.types.utils.TypeConversions.{fromDataTypeToLegacyInfo, fromLegacyInfoToDataType}
import scala.collection.JavaConversions._
@@ -99,13 +98,16 @@ class StreamExecTableSourceScan(
isStreamTable = true,
None)
+ val inputDataType = fromLegacyInfoToDataType(inputTransform.getOutputType)
+ val producedDataType = tableSource.getProducedDataType
+ val producedTypeInfo = fromDataTypeToLegacyInfo(producedDataType)
+
// check that declared and actual type of table source DataStream are identical
- if (createInternalTypeFromTypeInfo(inputTransform.getOutputType) !=
- createInternalTypeFromTypeInfo(tableSource.getReturnType)) {
+ if (inputDataType != producedDataType) {
throw new TableException(s"TableSource of type ${tableSource.getClass.getCanonicalName} " +
- s"returned a DataStream of type ${inputTransform.getOutputType} that does not match with " +
- s"the type ${tableSource.getReturnType} declared by the TableSource.getReturnType() " +
- s"method. Please validate the implementation of the TableSource.")
+ s"returned a DataStream of data type $producedDataType that does not match with the " +
+ s"data type $producedDataType declared by the TableSource.getProducedDataType() method. " +
+ s"Please validate the implementation of the TableSource.")
}
// get expression to extract rowtime attribute
@@ -131,7 +133,7 @@ class StreamExecTableSourceScan(
ctx,
inputTransform.asInstanceOf[StreamTransformation[Any]],
fieldIndexes,
- tableSource.getReturnType,
+ producedTypeInfo,
getRowType,
getTable.getQualifiedName,
config,
@@ -177,7 +179,7 @@ class StreamExecTableSourceScan(
None)
ScanUtil.hasTimeAttributeField(fieldIndexes) ||
ScanUtil.needsConversion(
- tableSource.getReturnType,
+ fromDataTypeToLegacyInfo(tableSource.getProducedDataType),
TypeExtractor.createTypeInfo(
tableSource, classOf[StreamTableSource[_]], tableSource.getClass, 0)
.getTypeClass.asInstanceOf[Class[_]])
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
index a4dc4c7..d39c7a9 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
@@ -32,7 +32,7 @@ import org.apache.flink.table.`type`.{InternalType, TypeConverters}
import org.apache.flink.table.api.{Types, ValidationException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.expressions._
-import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
+import org.apache.flink.table.types.utils.TypeConversions.{fromDataTypeToLegacyInfo, fromLegacyInfoToDataType}
import scala.collection.JavaConversions._
@@ -113,7 +113,7 @@ object TableSourceUtil {
}
idx
}
- val inputType = tableSource.getReturnType
+ val inputType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
// ensure that only one field is mapped to an atomic type
if (!inputType.isInstanceOf[CompositeType[_]] && mapping.count(_ >= 0) > 1) {
@@ -315,7 +315,7 @@ object TableSourceUtil {
fieldName: String,
tableSource: TableSource[_]): (String, Int, TypeInformation[_]) = {
- val returnType = tableSource.getReturnType
+ val returnType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
/** Look up a field by name in a type information */
def lookupField(fieldName: String, failMsg: String): (String, Int, TypeInformation[_]) = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
index 862523b..7f4a5fb 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
@@ -29,6 +29,7 @@ import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.plan.nodes.PhysicalTableSourceScan
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.sources._
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
import org.apache.flink.types.Row
/** Flink RelNode to read data from an external source defined by a [[BatchTableSource]]. */
@@ -91,11 +92,14 @@ class BatchTableSourceScan(
val inputDataSet = tableSource.getDataSet(tableEnv.execEnv).asInstanceOf[DataSet[Any]]
val outputSchema = new RowSchema(this.getRowType)
+ val inputDataType = fromLegacyInfoToDataType(inputDataSet.getType)
+ val producedDataType = tableSource.getProducedDataType
+
// check that declared and actual type of table source DataSet are identical
- if (inputDataSet.getType != tableSource.getReturnType) {
+ if (inputDataType != producedDataType) {
throw new TableException(s"TableSource of type ${tableSource.getClass.getCanonicalName} " +
- s"returned a DataSet of type ${inputDataSet.getType} that does not match with the " +
- s"type ${tableSource.getReturnType} declared by the TableSource.getReturnType() method. " +
+ s"returned a DataSet of data type $producedDataType that does not match with the " +
+ s"data type $producedDataType declared by the TableSource.getProducedDataType() method. " +
s"Please validate the implementation of the TableSource.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
index 20e2234..38216ba 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
@@ -33,6 +33,7 @@ import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.table.sources._
import org.apache.flink.table.sources.wmstrategies.{PeriodicWatermarkAssigner, PreserveWatermarks, PunctuatedWatermarkAssigner}
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
/** Flink RelNode to read data from an external source defined by a [[StreamTableSource]]. */
@@ -95,11 +96,14 @@ class StreamTableSourceScan(
val inputDataStream = tableSource.getDataStream(tableEnv.execEnv).asInstanceOf[DataStream[Any]]
val outputSchema = new RowSchema(this.getRowType)
+ val inputDataType = fromLegacyInfoToDataType(inputDataStream.getType)
+ val producedDataType = tableSource.getProducedDataType
+
// check that declared and actual type of table source DataStream are identical
- if (inputDataStream.getType != tableSource.getReturnType) {
+ if (inputDataType != producedDataType) {
throw new TableException(s"TableSource of type ${tableSource.getClass.getCanonicalName} " +
- s"returned a DataStream of type ${inputDataStream.getType} that does not match with the " +
- s"type ${tableSource.getReturnType} declared by the TableSource.getReturnType() method. " +
+ s"returned a DataStream of data type $producedDataType that does not match with the " +
+ s"data type $producedDataType declared by the TableSource.getProducedDataType() method. " +
s"Please validate the implementation of the TableSource.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
index 375b295..4173188 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
@@ -32,6 +32,7 @@ import org.apache.flink.api.common.typeutils.CompositeType
import org.apache.flink.table.api.{TableException, Types, ValidationException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.expressions.{Cast, PlannerExpression, PlannerResolvedFieldReference, ResolvedFieldReference}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import scala.collection.JavaConverters._
@@ -95,9 +96,10 @@ object TableSourceUtil {
mappedFieldCnt += 1
}
// ensure that only one field is mapped to an atomic type
- if (!tableSource.getReturnType.isInstanceOf[CompositeType[_]] && mappedFieldCnt > 1) {
+ val producedType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
+ if (!producedType.isInstanceOf[CompositeType[_]] && mappedFieldCnt > 1) {
throw new ValidationException(
- s"More than one table field matched to atomic input type ${tableSource.getReturnType}.")
+ s"More than one table field matched to atomic input type $producedType.")
}
// validate rowtime attributes
@@ -172,7 +174,7 @@ object TableSourceUtil {
tableSource: TableSource[_],
isStreamTable: Boolean,
selectedFields: Option[Array[Int]]): Array[Int] = {
- val inputType = tableSource.getReturnType
+ val inputType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
val tableSchema = tableSource.getTableSchema
// get names of selected fields
@@ -465,7 +467,7 @@ object TableSourceUtil {
fieldName: String,
tableSource: TableSource[_]): (String, Int, TypeInformation[_]) = {
- val returnType = tableSource.getReturnType
+ val returnType = fromDataTypeToLegacyInfo(tableSource.getProducedDataType)
/** Look up a field by name in a type information */
def lookupField(fieldName: String, failMsg: String): (String, Int, TypeInformation[_]) = {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
index 8787e94..ce50720 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
@@ -22,7 +22,9 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.types.Row;
import java.util.HashMap;
import java.util.Map;
@@ -265,18 +267,18 @@ public class CatalogStructureBuilder {
}
}
- private static class TestTable extends ConnectorCatalogTable<Object, Object> {
+ private static class TestTable extends ConnectorCatalogTable<Row, Row> {
private final String fullyQualifiedPath;
- private static final StreamTableSource<Object> tableSource = new StreamTableSource<Object>() {
+ private static final StreamTableSource<Row> tableSource = new StreamTableSource<Row>() {
@Override
- public DataStream<Object> getDataStream(StreamExecutionEnvironment execEnv) {
+ public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
return null;
}
@Override
- public TypeInformation<Object> getReturnType() {
- return null;
+ public TypeInformation<Row> getReturnType() {
+ return Types.ROW();
}
@Override