sunjincheng created FLINK-10875:
-----------------------------------
Summary: Add `toTableWithTimestamp` method in
`DataStreamConversions`
Key: FLINK-10875
URL: https://issues.apache.org/jira/browse/FLINK-10875
Project: Flink
Issue Type: Improvement
Components: Table API & SQL
Reporter: sunjincheng
Assignee: sunjincheng
Fix For: 1.7.1
Currently we convert a `DataStream` to a `Table` by
`DataStreamConversions#toTable`, e.g.:
{code:java}
// Without TimeAttribute
...
val stream = env.fromCollection(...)
val tab = stream.toTable(tEnv, 'a, 'b, 'c)
val result = tab.select('a, 'b)
....
// With TimeAttribute
...
val stream = env.fromCollection(...).assignTimestampsAndWatermarks(...)
val tab = stream.toTable(tEnv, 'a, 'b, 'c, 'ts.rowtime)
val result = tab.window(Session withGap 5.milli on 'ts as 'w)
...{code}
I think the fieldNames parameter in the `toTable` method is reasonable in the
conversion without the time attribute, because the fieldNames will actually
correspond to the fields of the physical table, but when applied to the
conversion with the time attribute, the time attribute column is silently added
to the table. This feeling is very Magical, so I recommend adding a method that
allows the user to display the time attribute added to the physical table:
`toTableWithTimestamp`, which is automatically named to the time attribute
column named by user input and TimeCharacteristic, eg:
{code:java}
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
...
val table = stream.toTableWithTimestamp(tEnv, 'count, 'size, 'name, 'ts)
.window(Tumble over 2.rows on 'ts as 'w)
...
{code}
In the example above the flink will mark `ts` ad a `RowtimeAttribute`.
What do you think ?
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
