[
https://issues.apache.org/jira/browse/FLINK-11220?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16729262#comment-16729262
]
sunjincheng commented on FLINK-11220:
-------------------------------------
Hi [~xccui] ,
Thanks for your explanations. Solving this problem does not mean we have to
support multiple watermarks. And I think the current logic of time-windowed
join is well. The purpose of this JIRA is to solve the actual needs of users, I
will explain the specific needs:
The SQL described above is a business scenario of order and payment. The actual
demand of the business is that the payment order of more than 1 hour is an
invalid order. Therefore, in the statistical inquiry, the order time and the
corresponding payment time must be queried for reporting, that is, the
orderTime and payTime columns must be selected. This is the actual business
requirement. We should support the user to write `select('orderTime, 'payTime)`
queries. So, The core requirement is that the following user SQL can work well：
{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}
Of course we can change the user's SQL, work around is as follows:
{code:java}
Orders...toTable(tEnv, 'orderId, 'orderTime.rowtime)
Payment...toTable(tEnv, 'orderId, 'payTime2, 'payTime.rowtime) // add a common
field for project.
SELECT orderTime, o.orderId, payTime2 //payTime --> payTime2
FROM Orders AS o JOIN Payment AS p
ON o.orderId = p.orderId AND
p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
{code}
But, I think It is difficult for users to imagine such a solution, because both
tables are defined as rows of rowtime, one can be projected, the other is not.
So we should improve our flink system, not let user change the SQL. What do you
think ?
Thanks,
Jincheng
> 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)
