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 2017/11/20 11:17:14 UTC
flink git commit: [FLINK-8096] [table] Fix time attribute
materialization when writing to TableSink
Repository: flink
Updated Branches:
refs/heads/master 51b5b53c7 -> 691c48a14
[FLINK-8096] [table] Fix time attribute materialization when writing to TableSink
This closes #5025.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/691c48a1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/691c48a1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/691c48a1
Branch: refs/heads/master
Commit: 691c48a14a138e4ff3ae330b7c9cfc0e596a9afa
Parents: 51b5b53
Author: Dian Fu <fu...@alibaba-inc.com>
Authored: Fri Nov 17 10:53:31 2017 +0800
Committer: twalthr <tw...@apache.org>
Committed: Mon Nov 20 10:52:27 2017 +0100
----------------------------------------------------------------------
.../table/api/StreamTableEnvironment.scala | 39 ++++++++++++--------
.../runtime/stream/TimeAttributesITCase.scala | 26 ++++++++++++-
2 files changed, 49 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/691c48a1/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
index e80acca..920da2e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -234,11 +234,12 @@ abstract class StreamTableEnvironment(
"UpsertStreamTableSink requires that Table has a full primary keys if it is updated.")
}
val outputType = sink.getOutputType
+ val resultType = getResultType(table.getRelNode, optimizedPlan)
// translate the Table into a DataStream and provide the type that the TableSink expects.
val result: DataStream[T] =
translate(
optimizedPlan,
- table.getRelNode.getRowType,
+ resultType,
streamQueryConfig,
withChangeFlag = true)(outputType)
// Give the DataStream to the TableSink to emit it.
@@ -254,11 +255,12 @@ abstract class StreamTableEnvironment(
"AppendStreamTableSink requires that Table has only insert changes.")
}
val outputType = sink.getOutputType
+ val resultType = getResultType(table.getRelNode, optimizedPlan)
// translate the Table into a DataStream and provide the type that the TableSink expects.
val result: DataStream[T] =
translate(
optimizedPlan,
- table.getRelNode.getRowType,
+ resultType,
streamQueryConfig,
withChangeFlag = false)(outputType)
// Give the DataStream to the TableSink to emit it.
@@ -727,19 +729,7 @@ abstract class StreamTableEnvironment(
val relNode = table.getRelNode
val dataStreamPlan = optimize(relNode, updatesAsRetraction)
- // zip original field names with optimized field types
- val fieldTypes = relNode.getRowType.getFieldList.asScala
- .zip(dataStreamPlan.getRowType.getFieldList.asScala)
- // get name of original plan and type of optimized plan
- .map(x => (x._1.getName, x._2.getType))
- // add field indexes
- .zipWithIndex
- // build new field types
- .map(x => new RelDataTypeFieldImpl(x._1._1, x._2, x._1._2))
-
- // build a record type from list of field types
- val rowType = new RelRecordType(
- fieldTypes.toList.asInstanceOf[List[RelDataTypeField]].asJava)
+ val rowType = getResultType(relNode, dataStreamPlan)
translate(dataStreamPlan, rowType, queryConfig, withChangeFlag)
}
@@ -852,6 +842,25 @@ abstract class StreamTableEnvironment(
}
/**
+ * Returns the record type of the optimized plan with field names of the logical plan.
+ */
+ private def getResultType(originRelNode: RelNode, optimizedPlan: RelNode): RelRecordType = {
+ // zip original field names with optimized field types
+ val fieldTypes = originRelNode.getRowType.getFieldList.asScala
+ .zip(optimizedPlan.getRowType.getFieldList.asScala)
+ // get name of original plan and type of optimized plan
+ .map(x => (x._1.getName, x._2.getType))
+ // add field indexes
+ .zipWithIndex
+ // build new field types
+ .map(x => new RelDataTypeFieldImpl(x._1._1, x._2, x._1._2))
+
+ // build a record type from list of field types
+ new RelRecordType(
+ fieldTypes.toList.asInstanceOf[List[RelDataTypeField]].asJava)
+ }
+
+ /**
* Returns the AST of the specified Table API and SQL queries and the execution plan to compute
* the result of the given [[Table]].
*
http://git-wip-us.apache.org/repos/asf/flink/blob/691c48a1/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala
index 5086601..a301354 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala
@@ -36,7 +36,7 @@ import org.apache.flink.table.api.{TableEnvironment, TableSchema, Types}
import org.apache.flink.table.expressions.{ExpressionParser, TimeIntervalUnit}
import org.apache.flink.table.runtime.stream.TimeAttributesITCase.{AtomicTimestampWithEqualWatermark, TestPojo, TimestampWithEqualWatermark, TimestampWithEqualWatermarkPojo}
import org.apache.flink.table.runtime.utils.StreamITCase
-import org.apache.flink.table.utils.TestTableSourceWithTime
+import org.apache.flink.table.utils.{MemoryTableSinkUtil, TestTableSourceWithTime}
import org.apache.flink.types.Row
import org.junit.Assert._
import org.junit.Test
@@ -179,6 +179,30 @@ class TimeAttributesITCase extends StreamingMultipleProgramsTestBase {
}
@Test
+ def testTableSink(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ MemoryTableSinkUtil.clear
+
+ val stream = env
+ .fromCollection(data)
+ .assignTimestampsAndWatermarks(new TimestampWithEqualWatermark())
+ stream.toTable(tEnv, 'rowtime.rowtime, 'int, 'double, 'float, 'bigdec, 'string)
+ .filter('rowtime.cast(Types.LONG) > 4)
+ .select('rowtime, 'rowtime.floor(TimeIntervalUnit.DAY), 'rowtime.ceil(TimeIntervalUnit.DAY))
+ .writeToSink(new MemoryTableSinkUtil.UnsafeMemoryAppendTableSink)
+
+ env.execute()
+
+ val expected = Seq(
+ "1970-01-01 00:00:00.007,1970-01-01 00:00:00.0,1970-01-02 00:00:00.0",
+ "1970-01-01 00:00:00.008,1970-01-01 00:00:00.0,1970-01-02 00:00:00.0",
+ "1970-01-01 00:00:00.016,1970-01-01 00:00:00.0,1970-01-02 00:00:00.0")
+ assertEquals(expected.sorted, MemoryTableSinkUtil.results.sorted)
+ }
+
+ @Test
def testTableFunction(): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)