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:23 UTC
[flink] 02/02: [FLINK-12254][table] Update TableSink 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 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))
}
}