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:21 UTC
[flink] branch master updated (5f5f02b -> 3351993)
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 5f5f02b [FLINK-12726][table-common] Fix ANY type serialization
new cedd980 [FLINK-12254][table] Update TableSource and related interfaces to new type system
new 3351993 [FLINK-12254][table] Update TableSink and related interfaces to new type system
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client/gateway/local/ExecutionContextTest.java | 4 +-
.../apache/flink/table/catalog/CatalogManager.java | 4 +-
.../flink/table/catalog/ConnectorCatalogTable.java | 3 +-
.../org/apache/flink/table/sinks/TableSink.java | 62 +++++++++++++++++++---
.../apache/flink/table/sinks/TableSinkBase.java | 3 ++
.../flink/table/sources/DefinedFieldMapping.java | 18 +++----
.../NestedFieldsProjectableTableSource.java | 10 ++--
.../table/sources/ProjectableTableSource.java | 8 +--
.../apache/flink/table/sources/TableSource.java | 34 +++++++++---
.../flink/table/codegen/SinkCodeGenerator.scala | 2 +-
.../flink/table/plan/nodes/calcite/Sink.scala | 12 ++---
.../table/plan/nodes/common/CommonLookupJoin.scala | 27 +++++-----
.../plan/nodes/physical/batch/BatchExecSink.scala | 12 ++---
.../physical/batch/BatchExecTableSourceScan.scala | 37 +++++++------
.../nodes/physical/stream/StreamExecSink.scala | 18 +++----
.../stream/StreamExecTableSourceScan.scala | 40 +++++++-------
.../flink/table/plan/schema/TableSinkTable.scala | 5 +-
.../flink/table/sinks/CollectTableSink.scala | 2 +-
.../flink/table/sinks/DataStreamTableSink.scala | 2 +-
.../flink/table/sources/TableSourceUtil.scala | 6 +--
.../table/runtime/utils/BatchTableEnvUtil.scala | 7 ++-
.../apache/flink/table/api/BatchTableEnvImpl.scala | 4 +-
.../flink/table/api/StreamTableEnvImpl.scala | 10 ++--
.../org/apache/flink/table/api/TableEnvImpl.scala | 12 ++---
.../plan/nodes/dataset/BatchTableSourceScan.scala | 10 ++--
.../nodes/datastream/StreamTableSourceScan.scala | 10 ++--
.../apache/flink/table/sinks/CsvTableSink.scala | 4 +-
.../flink/table/sources/TableSourceUtil.scala | 10 ++--
.../table/catalog/CatalogStructureBuilder.java | 12 +++--
.../table/utils/MemoryTableSourceSinkUtil.scala | 6 +--
30 files changed, 244 insertions(+), 150 deletions(-)
[flink] 01/02: [FLINK-12254][table] Update TableSource and related
interfaces to new type system
Posted by tw...@apache.org.
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
[flink] 02/02: [FLINK-12254][table] Update TableSink and related
interfaces to new type system
Posted by tw...@apache.org.
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 33519930a19735d531cd70d0fe32bf531a90ff0b
Author: Timo Walther <tw...@apache.org>
AuthorDate: Fri May 31 17:16:27 2019 +0200
[FLINK-12254][table] Update TableSink and related interfaces to new type system
This closes #8596.
---
.../client/gateway/local/ExecutionContextTest.java | 4 +-
.../apache/flink/table/catalog/CatalogManager.java | 4 +-
.../flink/table/catalog/ConnectorCatalogTable.java | 3 +-
.../org/apache/flink/table/sinks/TableSink.java | 62 +++++++++++++++++++---
.../apache/flink/table/sinks/TableSinkBase.java | 3 ++
.../apache/flink/table/sources/TableSource.java | 4 +-
.../flink/table/codegen/SinkCodeGenerator.scala | 2 +-
.../flink/table/plan/nodes/calcite/Sink.scala | 12 ++---
.../plan/nodes/physical/batch/BatchExecSink.scala | 12 ++---
.../nodes/physical/stream/StreamExecSink.scala | 18 +++----
.../flink/table/plan/schema/TableSinkTable.scala | 5 +-
.../flink/table/sinks/CollectTableSink.scala | 2 +-
.../flink/table/sinks/DataStreamTableSink.scala | 2 +-
.../table/runtime/utils/BatchTableEnvUtil.scala | 7 ++-
.../apache/flink/table/api/BatchTableEnvImpl.scala | 4 +-
.../flink/table/api/StreamTableEnvImpl.scala | 10 ++--
.../org/apache/flink/table/api/TableEnvImpl.scala | 12 ++---
.../apache/flink/table/sinks/CsvTableSink.scala | 4 +-
.../table/utils/MemoryTableSourceSinkUtil.scala | 6 +--
19 files changed, 113 insertions(+), 63 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
index 5137f63..bcdd991 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
@@ -124,11 +124,11 @@ public class ExecutionContextTest {
assertArrayEquals(
new String[]{"BooleanField", "StringField"},
- sinks.get("TableSourceSink").getFieldNames());
+ sinks.get("TableSourceSink").getTableSchema().getFieldNames());
assertArrayEquals(
new TypeInformation[]{Types.BOOLEAN(), Types.STRING()},
- sinks.get("TableSourceSink").getFieldTypes());
+ sinks.get("TableSourceSink").getTableSchema().getFieldTypes());
final TableEnvironment tableEnv = context.createEnvironmentInstance().getTableEnvironment();
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index 5704043..c965a98 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -383,8 +383,8 @@ public class CatalogManager {
if (externalTable.isTableSource()) {
return TableFactoryUtil.findAndCreateTableSource(externalTable).getTableSchema();
} else {
- TableSink tableSink = TableFactoryUtil.findAndCreateTableSink(externalTable);
- return new TableSchema(tableSink.getFieldNames(), tableSink.getFieldTypes());
+ TableSink<?> tableSink = TableFactoryUtil.findAndCreateTableSink(externalTable);
+ return tableSink.getTableSchema();
}
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
index ce86084..84b6e0d 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
@@ -57,8 +57,7 @@ public class ConnectorCatalogTable<T1, T2> extends AbstractCatalogTable {
}
public static <T2> ConnectorCatalogTable sink(TableSink<T2> sink, boolean isBatch) {
- TableSchema tableSchema = new TableSchema(sink.getFieldNames(), sink.getFieldTypes());
- return new ConnectorCatalogTable<>(null, sink, tableSchema, isBatch);
+ return new ConnectorCatalogTable<>(null, sink, sink.getTableSchema(), isBatch);
}
public static <T1, T2> ConnectorCatalogTable sourceAndSink(
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSink.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSink.java
index 4b13cb5..f8e003c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSink.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSink.java
@@ -20,6 +20,12 @@ package org.apache.flink.table.sinks;
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 static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType;
/**
* A {@link TableSink} specifies how to emit a table to an external
@@ -33,23 +39,59 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
public interface TableSink<T> {
/**
- * Returns the type expected by this {@link TableSink}.
+ * Returns the data type consumed by this {@link TableSink}.
*
- * <p>This type should depend on the types returned by {@link TableSink#getFieldNames()}.
+ * @return The data type expected by this {@link TableSink}.
+ */
+ default DataType getConsumedDataType() {
+ final TypeInformation<T> legacyType = getOutputType();
+ if (legacyType == null) {
+ throw new TableException("Table sink does not implement a consumed data type.");
+ }
+ return fromLegacyInfoToDataType(legacyType);
+ }
+
+ /**
+ * @deprecated This method will be removed in future versions as it uses the old type system. It
+ * is recommended to use {@link #getConsumedDataType()} 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.
+ */
+ @Deprecated
+ default TypeInformation<T> getOutputType() {
+ return null;
+ }
+
+ /**
+ * Returns the schema of the consumed table.
*
- * @return The type expected by this {@link TableSink}.
+ * @return The {@link TableSchema} of the consumed table.
*/
- TypeInformation<T> getOutputType();
+ default TableSchema getTableSchema() {
+ final String[] fieldNames = getFieldNames();
+ final TypeInformation[] legacyFieldTypes = getFieldTypes();
+ if (fieldNames == null || legacyFieldTypes == null) {
+ throw new TableException("Table sink does not implement a table schema.");
+ }
+ return new TableSchema(fieldNames, legacyFieldTypes);
+ }
/**
- * Returns the names of the table fields.
+ * @deprecated Use the field names of {@link #getTableSchema()} instead.
*/
- String[] getFieldNames();
+ @Deprecated
+ default String[] getFieldNames() {
+ return null;
+ }
/**
- * Returns the types of the table fields.
+ * @deprecated Use the field types of {@link #getTableSchema()} instead.
*/
- TypeInformation<?>[] getFieldTypes();
+ @Deprecated
+ default TypeInformation<?>[] getFieldTypes() {
+ return null;
+ }
/**
* Returns a copy of this {@link TableSink} configured with the field names and types of the
@@ -59,6 +101,10 @@ public interface TableSink<T> {
* @param fieldTypes The field types of the table to emit.
* @return A copy of this {@link TableSink} configured with the field names and types of the
* table to emit.
+ *
+ * @deprecated This method will be dropped in future versions. It is recommended to pass a
+ * static schema when instantiating the sink instead.
*/
+ @Deprecated
TableSink<T> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSinkBase.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSinkBase.java
index 830bf1c..3bf9f3f 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSinkBase.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sinks/TableSinkBase.java
@@ -42,6 +42,7 @@ public abstract class TableSinkBase<T> implements TableSink<T> {
/**
* Returns the field names of the table to emit.
*/
+ @Override
public String[] getFieldNames() {
if (fieldNames.isPresent()) {
return fieldNames.get();
@@ -54,6 +55,7 @@ public abstract class TableSinkBase<T> implements TableSink<T> {
/**
* Returns the field types of the table to emit.
*/
+ @Override
public TypeInformation<?>[] getFieldTypes() {
if (fieldTypes.isPresent()) {
return fieldTypes.get();
@@ -72,6 +74,7 @@ public abstract class TableSinkBase<T> implements TableSink<T> {
* @return A copy of this {@link TableSink} configured with the field names and types of the
* table to emit.
*/
+ @Override
public final TableSink<T> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
final TableSinkBase<T> configuredSink = this.copy();
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 bf379f6..386240b 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
@@ -46,7 +46,6 @@ public interface TableSource<T> {
/**
* 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 data type of the returned {@code DataSet} or {@code DataStream}.
*/
@@ -55,7 +54,7 @@ public interface TableSource<T> {
if (legacyType == null) {
throw new TableException("Table source does not implement a produced data type.");
}
- return fromLegacyInfoToDataType(getReturnType());
+ return fromLegacyInfoToDataType(legacyType);
}
/**
@@ -66,7 +65,6 @@ public interface TableSource<T> {
* for more information.
*/
@Deprecated
- @SuppressWarnings("unchecked")
default TypeInformation<T> getReturnType() {
return null;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
index 5ec3a38..6006ec1 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
@@ -44,7 +44,7 @@ object SinkCodeGenerator {
try {
sink match {
// DataStreamTableSink has no generic class, so we need get the type to get type class.
- case sink: DataStreamTableSink[_] => sink.getOutputType.getTypeClass
+ case sink: DataStreamTableSink[_] => sink.getConsumedDataType.getConversionClass
case _ => TypeExtractor.createTypeInfo(sink, classOf[TableSink[_]], sink.getClass, 0)
.getTypeClass.asInstanceOf[Class[_]]
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/calcite/Sink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/calcite/Sink.scala
index 29623c3..9d0dffa 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/calcite/Sink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/calcite/Sink.scala
@@ -18,13 +18,13 @@
package org.apache.flink.table.plan.nodes.calcite
-import org.apache.flink.table.`type`.TypeConverters
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.sinks.TableSink
-
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.flink.table.`type`.TypeConverters
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.sinks.TableSink
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
/**
* Relational expression that writes out data of input node into a [[TableSink]].
@@ -45,7 +45,7 @@ abstract class Sink(
override def deriveRowType(): RelDataType = {
val typeFactory = getCluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
- val outputType = sink.getOutputType
+ val outputType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
val internalType = TypeConverters.createInternalTypeFromTypeInfo(outputType)
typeFactory.createTypeFromInternalType(internalType, isNullable = true)
}
@@ -53,7 +53,7 @@ abstract class Sink(
override def explainTerms(pw: RelWriter): RelWriter = {
super.explainTerms(pw)
.itemIf("name", sinkName, sinkName != null)
- .item("fields", sink.getFieldNames.mkString(", "))
+ .item("fields", sink.getTableSchema.getFieldNames.mkString(", "))
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala
index 90e9a6e..a6f5dcb 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala
@@ -18,6 +18,10 @@
package org.apache.flink.table.plan.nodes.physical.batch
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
import org.apache.flink.runtime.operators.DamBehavior
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.transformations.{OneInputTransformation, StreamTransformation}
@@ -28,13 +32,9 @@ import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.plan.nodes.calcite.Sink
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.sinks.{BatchTableSink, DataStreamTableSink, TableSink}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.BaseRowTypeInfo
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-
-import java.util
-
import scala.collection.JavaConversions._
/**
@@ -98,7 +98,7 @@ class BatchExecSink[T](
private def translateToStreamTransformation(
withChangeFlag: Boolean,
tableEnv: BatchTableEnvironment): StreamTransformation[T] = {
- val resultType = sink.getOutputType
+ val resultType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
TableEnvironment.validateType(resultType)
val inputNode = getInputNodes.get(0)
inputNode match {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala
index fbfb776..d39b996 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala
@@ -18,26 +18,26 @@
package org.apache.flink.table.plan.nodes.physical.stream
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.transformations.{OneInputTransformation, StreamTransformation}
+import org.apache.flink.table.`type`.InternalTypes
import org.apache.flink.table.api.{StreamTableEnvironment, Table, TableException}
import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext}
import org.apache.flink.table.codegen.SinkCodeGenerator.{extractTableSinkTypeClass, generateRowConverterOperator}
+import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext}
import org.apache.flink.table.dataformat.BaseRow
+import org.apache.flink.table.plan.`trait`.{AccMode, AccModeTraitDef}
import org.apache.flink.table.plan.nodes.calcite.Sink
import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
-import org.apache.flink.table.plan.`trait`.{AccMode, AccModeTraitDef}
import org.apache.flink.table.plan.util.UpdatingPlanChecker
import org.apache.flink.table.sinks._
-import org.apache.flink.table.`type`.InternalTypes
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.BaseRowTypeInfo
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-
-import java.util
-
import scala.collection.JavaConversions._
/**
@@ -178,7 +178,7 @@ class StreamExecSink[T](
} else {
parTransformation.getOutputType
}
- val resultType = sink.getOutputType
+ val resultType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
val typeClass = extractTableSinkTypeClass(sink)
if (CodeGenUtils.isInternalClass(typeClass, resultType)) {
parTransformation.asInstanceOf[StreamTransformation[T]]
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
index 63d8204..fe02425 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
@@ -35,8 +35,9 @@ class TableSinkTable[T](
override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
val flinkTypeFactory = typeFactory.asInstanceOf[FlinkTypeFactory]
- val fieldTypes = tableSink.getFieldTypes.map(TypeConverters.createInternalTypeFromTypeInfo)
- flinkTypeFactory.buildRelDataType(tableSink.getFieldNames, fieldTypes)
+ val fieldTypes = tableSink.getTableSchema.getFieldTypes
+ .map(TypeConverters.createInternalTypeFromTypeInfo)
+ flinkTypeFactory.buildRelDataType(tableSink.getTableSchema.getFieldNames, fieldTypes)
}
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
index 3cc3903..8c50f57 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
@@ -51,7 +51,7 @@ class CollectTableSink[T](produceOutputType: (Array[TypeInformation[_]] => TypeI
}
override def getOutputType: TypeInformation[T] = {
- produceOutputType(getFieldTypes)
+ produceOutputType(getTableSchema.getFieldTypes)
}
def init(typeSerializer: TypeSerializer[T], id: String): Unit = {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala
index c4a308d..c421706 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala
@@ -44,7 +44,7 @@ class DataStreamTableSink[T](
/**
* Return the type expected by this [[TableSink]].
*
- * This type should depend on the types returned by [[getFieldNames]].
+ * This type should depend on the types returned by [[getTableSchema]].
*
* @return The type expected by this [[TableSink]].
*/
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/BatchTableEnvUtil.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/BatchTableEnvUtil.scala
index fe93155..2ac5a96 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/BatchTableEnvUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/utils/BatchTableEnvUtil.scala
@@ -29,9 +29,10 @@ import org.apache.flink.table.plan.schema.DataStreamTable
import org.apache.flink.table.plan.stats.FlinkStatistic
import org.apache.flink.table.sinks.CollectTableSink
import org.apache.flink.util.AbstractID
-
import _root_.java.util.{ArrayList => JArrayList}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
+
import _root_.scala.collection.JavaConversions._
import _root_.scala.collection.JavaConverters._
@@ -42,7 +43,9 @@ object BatchTableEnvUtil {
table: Table,
sink: CollectTableSink[T],
jobName: Option[String]): Seq[T] = {
- val typeSerializer = sink.getOutputType.createSerializer(tEnv.streamEnv.getConfig)
+ val typeSerializer = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
+ .createSerializer(tEnv.streamEnv.getConfig)
val id = new AbstractID().toString
sink.init(typeSerializer.asInstanceOf[TypeSerializer[T]], id)
tEnv.writeToSink(table, sink)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
index 66a1d83..0d6e133 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
@@ -41,6 +41,7 @@ import org.apache.flink.table.plan.schema._
import org.apache.flink.table.runtime.MapRunner
import org.apache.flink.table.sinks._
import org.apache.flink.table.sources.{BatchTableSource, TableSource, TableSourceUtil}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.FieldInfoUtils.{calculateTableSchema, getFieldsInfo, validateInputTypeInfo}
import org.apache.flink.types.Row
@@ -108,7 +109,8 @@ abstract class BatchTableEnvImpl(
sink match {
case batchSink: BatchTableSink[T] =>
- val outputType = sink.getOutputType
+ val outputType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
// translate the Table into a DataSet and provide the type that the TableSink expects.
val result: DataSet[T] = translate(table, batchQueryConfig)(outputType)
// Give the DataSet to the TableSink to emit it.
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
index 32b3f8c..d8f3edc 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
@@ -52,6 +52,7 @@ import org.apache.flink.table.runtime.{CRowMapRunner, OutputRowtimeProcessFuncti
import org.apache.flink.table.sinks._
import org.apache.flink.table.sources.{StreamTableSource, TableSource, TableSourceUtil}
import org.apache.flink.table.typeutils.FieldInfoUtils.{calculateTableSchema, getFieldsInfo, isReferenceByPosition}
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import org.apache.flink.table.typeutils.{TimeIndicatorTypeInfo, TypeCheckUtils}
import _root_.scala.collection.JavaConverters._
@@ -139,7 +140,8 @@ abstract class StreamTableEnvImpl(
case retractSink: RetractStreamTableSink[_] =>
// retraction sink can always be used
- val outputType = sink.getOutputType
+ val outputType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
// translate the Table into a DataStream and provide the type that the TableSink expects.
val result: DataStream[T] =
translate(
@@ -166,7 +168,8 @@ abstract class StreamTableEnvImpl(
case None if !isAppendOnlyTable => throw new TableException(
"UpsertStreamTableSink requires that Table has full primary keys if it is updated.")
}
- val outputType = sink.getOutputType
+ val outputType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
val resultType = getResultType(table.getRelNode, optimizedPlan)
// translate the Table into a DataStream and provide the type that the TableSink expects.
val result: DataStream[T] =
@@ -187,7 +190,8 @@ abstract class StreamTableEnvImpl(
throw new TableException(
"AppendStreamTableSink requires that Table has only insert changes.")
}
- val outputType = sink.getOutputType
+ val outputType = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
val resultType = getResultType(table.getRelNode, optimizedPlan)
// translate the Table into a DataStream and provide the type that the TableSink expects.
val result: DataStream[T] =
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
index f281829..b9ae912 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
@@ -455,15 +455,9 @@ abstract class TableEnvImpl(
override def registerTableSink(name: String, configuredSink: TableSink[_]): Unit = {
// validate
- if (configuredSink.getFieldNames == null || configuredSink.getFieldTypes == null) {
- throw new TableException("Table sink is not configured.")
- }
- if (configuredSink.getFieldNames.length == 0) {
+ if (configuredSink.getTableSchema.getFieldNames.length == 0) {
throw new TableException("Field names must not be empty.")
}
- if (configuredSink.getFieldNames.length != configuredSink.getFieldTypes.length) {
- throw new TableException("Same number of field names and types required.")
- }
validateTableSink(configuredSink)
registerTableSinkInternal(name, configuredSink)
@@ -692,13 +686,13 @@ abstract class TableEnvImpl(
case Some(tableSink) =>
// validate schema of source table and table sink
val srcFieldTypes = table.getSchema.getFieldTypes
- val sinkFieldTypes = tableSink.getFieldTypes
+ val sinkFieldTypes = tableSink.getTableSchema.getFieldTypes
if (srcFieldTypes.length != sinkFieldTypes.length ||
srcFieldTypes.zip(sinkFieldTypes).exists { case (srcF, snkF) => srcF != snkF }) {
val srcFieldNames = table.getSchema.getFieldNames
- val sinkFieldNames = tableSink.getFieldNames
+ val sinkFieldNames = tableSink.getTableSchema.getFieldNames
// format table and table sink schema strings
val srcSchema = srcFieldNames.zip(srcFieldTypes)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/CsvTableSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/CsvTableSink.scala
index 74efcf3..8037f4d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/CsvTableSink.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/CsvTableSink.scala
@@ -80,7 +80,7 @@ class CsvTableSink(
sink.setParallelism(numFiles.get)
}
- sink.name(TableConnectorUtils.generateRuntimeName(this.getClass, getFieldNames))
+ sink.name(TableConnectorUtils.generateRuntimeName(this.getClass, getTableSchema.getFieldNames))
}
override def emitDataStream(dataStream: DataStream[Row]): Unit = {
@@ -99,7 +99,7 @@ class CsvTableSink(
sink.setParallelism(numFiles.get)
}
- sink.name(TableConnectorUtils.generateRuntimeName(this.getClass, getFieldNames))
+ sink.name(TableConnectorUtils.generateRuntimeName(this.getClass, getTableSchema.getFieldNames))
}
override protected def copy: TableSinkBase[Row] = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MemoryTableSourceSinkUtil.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MemoryTableSourceSinkUtil.scala
index 2ee051b..37342c9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MemoryTableSourceSinkUtil.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MemoryTableSourceSinkUtil.scala
@@ -104,7 +104,7 @@ object MemoryTableSourceSinkUtil {
with AppendStreamTableSink[Row] {
override def getOutputType: TypeInformation[Row] = {
- new RowTypeInfo(getFieldTypes, getFieldNames)
+ new RowTypeInfo(getTableSchema.getFieldTypes, getTableSchema.getFieldNames)
}
override protected def copy: TableSinkBase[Row] = {
@@ -114,7 +114,7 @@ object MemoryTableSourceSinkUtil {
override def emitDataSet(dataSet: DataSet[Row]): Unit = {
dataSet
.output(new MemoryCollectionOutputFormat)
- .name(TableConnectorUtils.generateRuntimeName(this.getClass, getFieldNames))
+ .name(TableConnectorUtils.generateRuntimeName(this.getClass, getTableSchema.getFieldNames))
}
override def emitDataStream(dataStream: DataStream[Row]): Unit = {
@@ -122,7 +122,7 @@ object MemoryTableSourceSinkUtil {
dataStream
.addSink(new MemoryAppendSink)
.setParallelism(inputParallelism)
- .name(TableConnectorUtils.generateRuntimeName(this.getClass, getFieldNames))
+ .name(TableConnectorUtils.generateRuntimeName(this.getClass, getTableSchema.getFieldNames))
}
}