[
https://issues.apache.org/jira/browse/FLINK-11220?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16729569#comment-16729569
]
sunjincheng commented on FLINK-11220:
-------------------------------------
Thanks for your reply [~jark] , [~xccui] ！
Yes, I think we should do our best for the convenience of users.
For improve this problem, I did not think deeply, an easy way is to unify the
materialization of RowtimeIndicator and ProctimeIndicator. This means that the
RowtimeIndicator type is changed to the SqlTimeTypeInfo.TIMESTAMP type during
the optimization phase.
{code:java}
optimize(..){
...
val planWithMaterializedTimeAttributes =
RelTimeIndicatorConverter.convert(decorPlan, getRelBuilder.getRexBuilder) //
materialize remaining proctime and rowtime indicators
...
}{code}
The advantage of this is that while solving the current problems, it also
unifies the materialization process of RowtimeIndicator and ProctimeIndicator.
The bad thing is that if user want do rowtime window aggregate after
toDataStream, user need to call the assignTimestampsAndWatermarks() method to
extract the timestamp of the StreamRecord and generate the watermark.
What do you think?
> Can not Select row time field in JOIN query
> -------------------------------------------
>
> Key: FLINK-11220
> URL: https://issues.apache.org/jira/browse/FLINK-11220
> Project: Flink
> Issue Type: Bug
> Components: Table API & SQL
> Affects Versions: 1.8.0
> Reporter: sunjincheng
> Priority: Major
>
> SQL:
> {code:java}
> Orders...toTable(tEnv, 'orderId, 'orderTime.rowtime)
> Payment...toTable(tEnv, 'orderId, 'payTime.rowtime)
> SELECT orderTime, o.orderId, payTime
> FROM Orders AS o JOIN Payment AS p
> ON o.orderId = p.orderId AND
> p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
> {code}
> Execption:
> {code:java}
> org.apache.flink.table.api.TableException: Found more than one rowtime field:
> [orderTime, payTime] in the table that should be converted to a DataStream.
> Please select the rowtime field that should be used as event-time timestamp
> for the DataStream by casting all other fields to TIMESTAMP.
> at
> org.apache.flink.table.api.StreamTableEnvironment.translate(StreamTableEnvironment.scala:906)
> {code}
> The reason for the error is that we have 2 time fields `orderTime` and
> `payTime`. I think we do not need throw the exception, and we can remove
> the logic of `plan.process(new OutputRowtimeProcessFunction[A](conversion,
> rowtimeFields.head.getIndex))`, if we want using the timestamp after
> toDataSteram, we should using `assignTimestampsAndWatermarks()`.
> What do you think ? [~twalthr] [~fhueske]
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
