[
https://issues.apache.org/jira/browse/FLINK-11220?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730189#comment-16730189
]
sunjincheng commented on FLINK-11220:
-------------------------------------
Hi [~xccui], unifies the materialization process of RowtimeIndicator and
ProctimeIndicator, does not destroy the logic of the current window-join, that
is, after window-join, we can still use RowtimeIndicator Fileds, which means
that the following SQL statements are still supported:
{code:java}
SELECT COUNT(*) FROM (
SELECT
o.orderTime,
p.payTime
FROM
Orders AS o JOIN Payment AS p ON
o.orderId = p.orderId AND
p.payTime BETWEEN orderTime - INTERVAL '1' HOUR AND orderTime + INTERVAL '1'
HOUR
)
GROUP BY TUMBLE(payTime, INTERVAL '15' MINUTE)
{code}
We only do the materialization when we call UDX call or insert into the Sink.
We need to call `assignTimestampsAndWatermarks() ` when perform Event-time
operation after toDataStream, such as window Aggregate:
{code:java}
...SQL...
var result = ...toAppendStream[Row] // toDataStream
result = result.assignTimestampsAndWatermarks(...) // we should Extract
timestamp
result.windowAll(TumblingEventTimeWindows.of(..)).aggregate(...){code}
I don't know if it has been explained clearly, welcome any feedback.
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)
