Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Wed, 19 Apr 2017 09:11:42 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13061149.1491231820000.315945.1492593102137@Atlassian.JIRA>
In-Reply-To: <JIRA.13061149.1491231820000@Atlassian.JIRA>
References: <JIRA.13061149.1491231820000@Atlassian.JIRA> <JIRA.13061149.1491231820659@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-6250) Distinct procTime with Rows
 boundaries
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Wed, 19 Apr 2017 09:11:46 -0000


    [ https://issues.apache.org/jira/browse/FLINK-6250?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15974357#comment-15974357 ] 

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

Github user huawei-flink commented on a diff in the pull request:

    https://github.com/apache/flink/pull/3732#discussion_r112151646
  
    --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeBoundedDistinctRowsOver.scala ---
    @@ -0,0 +1,238 @@
    +/*
    + * 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 implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +package org.apache.flink.table.runtime.aggregate
    +
    +import java.util
    +
    +import org.apache.flink.configuration.Configuration
    +import org.apache.flink.streaming.api.functions.ProcessFunction
    +import org.apache.flink.types.Row
    +import org.apache.flink.util.{Collector, Preconditions}
    +import org.apache.flink.api.common.state.ValueStateDescriptor
    +import org.apache.flink.api.java.typeutils.RowTypeInfo
    +import org.apache.flink.api.common.state.ValueState
    +import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
    +import org.apache.flink.api.common.state.MapState
    +import org.apache.flink.api.common.state.MapStateDescriptor
    +import org.apache.flink.api.common.typeinfo.TypeInformation
    +import org.apache.flink.api.java.typeutils.ListTypeInfo
    +import java.util.{List => JList}
    +
    +import org.apache.flink.api.common.typeinfo.BasicTypeInfo
    +
    +class ProcTimeBoundedDistinctRowsOver(
    +  private val aggregates: Array[AggregateFunction[_]],
    +  private val aggFields: Array[Array[Int]],
    +  private val distinctAggsFlag: Array[Boolean],
    +  private val precedingOffset: Long,
    +  private val forwardedFieldCount: Int,
    +  private val aggregatesTypeInfo: RowTypeInfo,
    +  private val inputType: TypeInformation[Row])
    +    extends ProcessFunction[Row, Row] {
    +
    +  Preconditions.checkNotNull(aggregates)
    +  Preconditions.checkNotNull(aggFields)
    +  Preconditions.checkNotNull(distinctAggsFlag)
    +  Preconditions.checkNotNull(distinctAggsFlag.length == aggregates.length)
    +  Preconditions.checkArgument(aggregates.length == aggFields.length)
    +  Preconditions.checkArgument(precedingOffset > 0)
    +
    +  private var accumulatorState: ValueState[Row] = _
    +  private var rowMapState: MapState[Long, JList[Row]] = _
    +  private var output: Row = _
    +  private var counterState: ValueState[Long] = _
    +  private var smallestTsState: ValueState[Long] = _
    +  private var distinctValueState: MapState[Any, Row] = _
    +
    +  override def open(config: Configuration) {
    +
    +    output = new Row(forwardedFieldCount + aggregates.length)
    +    // We keep the elements received in a Map state keyed
    +    // by the ingestion time in the operator.
    +    // we also keep counter of processed elements
    +    // and timestamp of oldest element
    +    val rowListTypeInfo: TypeInformation[JList[Row]] =
    +      new ListTypeInfo[Row](inputType).asInstanceOf[TypeInformation[JList[Row]]]
    +
    +    val mapStateDescriptor: MapStateDescriptor[Long, JList[Row]] =
    +      new MapStateDescriptor[Long, JList[Row]]("windowBufferMapState",
    +        BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]], rowListTypeInfo)
    +    rowMapState = getRuntimeContext.getMapState(mapStateDescriptor)
    +
    +    val aggregationStateDescriptor: ValueStateDescriptor[Row] =
    +      new ValueStateDescriptor[Row]("aggregationState", aggregatesTypeInfo)
    +    accumulatorState = getRuntimeContext.getState(aggregationStateDescriptor)
    +
    +    val processedCountDescriptor : ValueStateDescriptor[Long] =
    +       new ValueStateDescriptor[Long]("processedCountState", classOf[Long])
    +    counterState = getRuntimeContext.getState(processedCountDescriptor)
    +
    +    val smallestTimestampDescriptor : ValueStateDescriptor[Long] =
    +       new ValueStateDescriptor[Long]("smallestTSState", classOf[Long])
    +    smallestTsState = getRuntimeContext.getState(smallestTimestampDescriptor)
    +    
    +    val distinctValDescriptor : MapStateDescriptor[Any, Row] =
    +      new MapStateDescriptor[Any, Row]("distinctValuesBufferMapState", classOf[Any], classOf[Row])
    +    distinctValueState = getRuntimeContext.getMapState(distinctValDescriptor)
    +  }
    +
    +  override def processElement(
    +    input: Row,
    +    ctx: ProcessFunction[Row, Row]#Context,
    +    out: Collector[Row]): Unit = {
    +
    +    val currentTime = ctx.timerService.currentProcessingTime
    +    var i = 0
    +
    +    // initialize state for the processed element
    +    var accumulators = accumulatorState.value
    +    if (accumulators == null) {
    +      accumulators = new Row(aggregates.length)
    +      while (i < aggregates.length) {
    +        accumulators.setField(i, aggregates(i).createAccumulator())
    +        i += 1
    +      }
    +    }
    +
    +    // get smallest timestamp
    +    var smallestTs = smallestTsState.value
    +    if (smallestTs == 0L) {
    +      smallestTs = currentTime
    +      smallestTsState.update(smallestTs)
    +    }
    +    // get previous counter value
    +    var counter = counterState.value
    +
    +    if (counter == precedingOffset) {
    +      val retractList = rowMapState.get(smallestTs)
    +
    +      // get oldest element beyond buffer size
    +      // and if oldest element exist, retract value
    +      var removeCounter :Integer = 0
    +      var distinctCounter : Integer = 0
    +      var retractVal : Object = null
    +      i = 0
    +      while (i < aggregates.length) {
    +        val accumulator = accumulators.getField(i).asInstanceOf[Accumulator]
    +        retractVal = retractList.get(0).getField(aggFields(i)(0))
    +        if(distinctAggsFlag(i)){
    +          distinctCounter += 1
    +          val counterRow = distinctValueState.get(retractVal)
    --- End diff --
    
    the map state takes Any as key, but I agree with you that I didn't test this aspect through


> Distinct procTime with Rows boundaries
> --------------------------------------
>
>                 Key: FLINK-6250
>                 URL: https://issues.apache.org/jira/browse/FLINK-6250
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Table API & SQL
>            Reporter: radu
>            Assignee: Stefano Bortoli
>
> Support proctime with rows boundaries
> Q1.1. `SELECT SUM( DISTINCT  b) OVER (ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM stream1`
> Q1.1. `SELECT COUNT(b), SUM( DISTINCT  b) OVER (ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM stream1`


--
This message was sent by Atlassian JIRA
(v6.3.15#6346)