[
https://issues.apache.org/jira/browse/FLINK-6091?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15981870#comment-15981870
]
ASF GitHub Bot commented on FLINK-6091:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3733#discussion_r113044668
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala
---
@@ -23,26 +23,28 @@ import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex._
import org.apache.flink.api.common.functions.{FlatMapFunction,
RichFlatMapFunction}
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.{CodeGenerator, GeneratedFunction}
import org.apache.flink.table.runtime.FlatMapRunner
+import org.apache.flink.table.runtime.types.CRowTypeInfo
import org.apache.flink.types.Row
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
trait CommonCalc {
- private[flink] def functionBody(
+ private[flink] def functionBody[T](
--- End diff --
I would follow a similar strategy as with the `TableEnvironment` sink
conversion for the `CommonX` classes.
The `CommonX` class has a method to generate the code for a function that
operates on `Row`. Each class that extends `CommonX` has a dedicated method to
instantiate the wrapper class (e..g, `MapRunner`, `FlatMapRunner`. ). The
runner classes use the generated function to operate on Rows. For streaming
operators, the runner classes unwrap the `Row` from `CRow` before calling the
generated code.
> Implement and turn on the retraction for aggregates
> ---------------------------------------------------
>
> Key: FLINK-6091
> URL: https://issues.apache.org/jira/browse/FLINK-6091
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Reporter: Shaoxuan Wang
> Assignee: Hequn Cheng
>
> Implement functions for generating and consuming retract messages for
> different aggregates.
> 1. add delete/add property to Row
> 2. implement functions for generating retract messages for unbounded groupBy
> 3. implement functions for handling retract messages for different aggregates.
> 4. handle retraction messages in CommonCorrelate and CommonCalc (retain
> Delete property).
> Note: Currently, only unbounded groupby generates retraction and it is
> working under unbounded and processing time mode. Hence, retraction is only
> supported for unbounded and processing time aggregations so far. We can add
> more retraction support later.
> supported now: unbounded groupby, unbounded and processing time over window
> unsupported now: group window, event time or bounded over window.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
