Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Tue, 19 Sep 2017 08:51:01 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13060634.1490953188000.144744.1505811061182@Atlassian.JIRA>
In-Reply-To: <JIRA.13060634.1490953188000@Atlassian.JIRA>
References: <JIRA.13060634.1490953188000@Atlassian.JIRA> <JIRA.13060634.1490953188313@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-6233) Support rowtime inner equi-join
 between two streams in the SQL API
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Tue, 19 Sep 2017 08:51:07 -0000


    [ https://issues.apache.org/jira/browse/FLINK-6233?page=3Dcom.atlassian=
.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=3D1617=
1333#comment-16171333 ]=20

ASF GitHub Bot commented on FLINK-6233:
---------------------------------------

Github user xccui commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4625#discussion_r139631978
 =20
    --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/t=
able/runtime/join/RowTimeBoundedStreamInnerJoin.scala ---
    @@ -0,0 +1,83 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + *     http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or imp=
lied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.flink.table.runtime.join
    +
    +import org.apache.flink.api.common.typeinfo.TypeInformation
    +import org.apache.flink.streaming.api.functions.co.CoProcessFunction
    +import org.apache.flink.table.runtime.types.CRow
    +import org.apache.flink.types.Row
    +
    +/**
    +  * The function to execute row(event) time bounded stream inner-join.
    +  */
    +class RowTimeBoundedStreamInnerJoin(
    +    leftLowerBound: Long,
    +    leftUpperBound: Long,
    +    allowedLateness: Long,
    +    leftType: TypeInformation[Row],
    +    rightType: TypeInformation[Row],
    +    genJoinFuncName: String,
    +    genJoinFuncCode: String,
    +    leftTimeIdx: Int,
    +    rightTimeIdx: Int)
    +    extends TimeBoundedStreamInnerJoin(
    +      leftLowerBound,
    +      leftUpperBound,
    +      allowedLateness,
    +      leftType,
    +      rightType,
    +      genJoinFuncName,
    +      genJoinFuncCode,
    +      leftTimeIdx,
    +      rightTimeIdx,
    +      JoinTimeIndicator.ROWTIME) {
    +
    +  override def checkRowOutOfDate(timeForRow: Long, watermark: Long) =
=3D {
    +    timeForRow <=3D watermark - allowedLateness
    +  }
    +
    +  override def updateOperatorTime(ctx: CoProcessFunction[CRow, CRow, C=
Row]#Context): Unit =3D {
    +    rightOperatorTime =3D
    +      if (ctx.timerService().currentWatermark() > 0) ctx.timerService(=
).currentWatermark()
    --- End diff --
   =20
    After checking the codes, I find out that the number of watermark usage=
s is even larger than that of their updates (i.e., each update is followed =
by at least one usage). Considering that the number of usage may increase i=
n the future, I suggest to keep this check in the update method. Of course,=
 it would be better if we can use a "safer" initial value for
     the watermark=F0=9F=98=84. What do you think?=20


> Support rowtime inner equi-join between two streams in the SQL API
> ------------------------------------------------------------------
>
>                 Key: FLINK-6233
>                 URL: https://issues.apache.org/jira/browse/FLINK-6233
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Table API & SQL
>            Reporter: hongyuhong
>            Assignee: Xingcan Cui
>
> The goal of this issue is to add support for inner equi-join on proc time=
 streams to the SQL interface.
> Queries similar to the following should be supported:
> {code}
> SELECT o.rowtime , o.productId, o.orderId, s.rowtime AS shipTime=20
> FROM Orders AS o=20
> JOIN Shipments AS s=20
> ON o.orderId =3D s.orderId=20
> AND o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL '1' HOUR;
> {code}
> The following restrictions should initially apply:
> * The join hint only support inner join
> * The ON clause should include equi-join condition
> * The time-condition {{o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERV=
AL '1' HOUR}} only can use rowtime that is a system attribute, the time con=
dition only support bounded time range like {{o.rowtime BETWEEN s.rowtime -=
 INTERVAL '1' HOUR AND s.rowtime + INTERVAL '1' HOUR}}, not support unbound=
ed like {{o.rowtime &lt; s.rowtime}} ,  and  should include both two stream=
's rowtime attribute, {{o.rowtime between rowtime () and rowtime () + 1}} s=
hould also not be supported.
> An row-time streams join will not be able to handle late data, because th=
is would mean in insert a row into a sorted order shift all other computati=
ons. This would be too expensive to maintain. Therefore, we will throw an e=
rror if a user tries to use an row-time stream join with late data handling=
.
> This issue includes:
> * Design of the DataStream operator to deal with stream join
> * Translation from Calcite's RelNode representation (LogicalJoin).=20


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)