title: Having an equivalent to HOP_START inside an aggregation primitive in Flink
body: <p>I'm trying to do an exponentially decaying moving average over a hopping window in Flink SQL. I need the have access to one of the borders of the window, the HOP_START in the following:</p>
  
  <pre class="lang-sql prettyprint-override"><code>    SELECT                                                                              
        lb_index one_key,
      -- I have access to this one:
        HOP_START(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND) start_time,
      -- Aggregation primitive:
        SUM(
          Y * EXP(TIMESTAMPDIFF(
            SECOND, 
            proctime, 
      -- This one throws:
            HOP_START(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND)
        )))
      FROM write_position                                                                
      GROUP BY lb_index, HOP(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND)
  </code></pre>
  
  <p>I'm getting the following stack trace:</p>
  
  <pre><code>11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using Aggregate(groupBy: (lb_index), window: (SlidingGroupWindow('w$, 'proctime, 5000.millis, 50.millis)), select: (lb_index, SUM($f2) AS Y, start('w$) AS w$start, end('w$) AS w$end, proctime('w$) AS w$proctime))
  11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using Calc(select: (lb_index, proctime, *(payload.Y, EXP(/(CAST(/INT(Reinterpret(-(HOP_START(PROCTIME(proctime), 50, 5000), PROCTIME(proctime))), 1000)), 1000))) AS $f2))
  11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using rel#459:DataStreamScan.DATASTREAM.true.Acc(table=[_DataStreamTable_0])
  Exception in thread "main" org.apache.flink.table.codegen.CodeGenException: Unsupported call: HOP_START 
  If you think this function should be supported, you can create an issue and start a discussion for it.
      at org.apache.flink.table.codegen.CodeGenerator$$anonfun$visitCall$3.apply(CodeGenerator.scala:1027)
      at org.apache.flink.table.codegen.CodeGenerator$$anonfun$visitCall$3.apply(CodeGenerator.scala:1027)
      at scala.Option.getOrElse(Option.scala:121)
      at org.apache.flink.table.codegen.CodeGenerator.visitCall(CodeGenerator.scala:1027)
      at org.apache.flink.table.codegen.CodeGenerator.visitCall(CodeGenerator.scala:66)
  </code></pre>
  
  <p>It does say is it unimplemented while it works outside the aggregating SUM. So that's what makes me think this is a scoping issue.</p>
  
  <p>Now, the thing is: I could transform this expression and do a final processing outside the aggregation, as exp(x+y) = exp(x)*exp(y); But I'm stuck with using TIMESTAMPDIFF (<a href="https://stackoverflow.com/questions/54655063/an-exponentially-decaying-moving-average-over-a-hopping-window-in-flink-sql-cas">which did wonders in my previous issue</a>). I have not found a way to cast TIME ATTRIBUTEs to NUMERIC types; also, I'm not comfortable exponentiating UNIX timestamps, even if I scale them down.</p>
  
  <p>Anyway, this work-around would be sort of clunky and there might me another way. I don't know how I could massage scopes in this SQL piece to still 'be' in the window scope and have the start time without throwing.</p>