[jira] [Created] (FLINK-7338) User Defined aggregation with constants causes error under in lowerbound over window extraction

["Stefano Bortoli (JIRA)" <ji...@apache.org>]

Stefano Bortoli created FLINK-7338:
--------------------------------------

             Summary: User Defined aggregation with constants causes error under in lowerbound over window extraction
                 Key: FLINK-7338
                 URL: https://issues.apache.org/jira/browse/FLINK-7338
             Project: Flink
          Issue Type: Bug
          Components: DataStream API
    Affects Versions: 1.3.1
            Reporter: Stefano Bortoli
            Priority: Critical


A user defined aggregation that passes a constant among the arguments causes a RuntimeException extracting the lower boundary over window.

[code]
val sqlQuery = "SELECT a, " +
      "  myAgg(a, CAST('1' as BIGINT)) "+
      "   OVER (ORDER BY proctime RANGE BETWEEN INTERVAL '30' SECOND 
               PRECEDING AND CURRENT ROW) " +
      "FROM MyTable"
[code]

The error is in the org.apache.flink.table.plan.nodes.OverAggregate.scala

we do : field count - lower bound index
--  which causes a -1 get, and subsequent RuntimeException. 
We should do: lower bound offset - field count to find the value in the constant array.

The code below should fix the problem.

[code]
private[flink] def getLowerBoundary(
    logicWindow: Window,
    overWindow: Group,
    input: RelNode): Long = {

    val ref: RexInputRef = overWindow.lowerBound.getOffset.asInstanceOf[RexInputRef]
    val lowerBoundIndex = ref.getIndex - input.getRowType.getFieldCount
    val lowerBound = logicWindow.constants.get(lowerBoundIndex).getValue2
    lowerBound match {
      case x: java.math.BigDecimal => x.asInstanceOf[java.math.BigDecimal].longValue()
      case _ => lowerBound.asInstanceOf[Long]
    }
  }
[code]



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