[
https://issues.apache.org/jira/browse/FLINK-4937?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15685073#comment-15685073
]
ASF GitHub Bot commented on FLINK-4937:
---------------------------------------
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/2792#discussion_r89012878
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateMapFunction.scala
---
@@ -30,9 +30,9 @@ class AggregateMapFunction[IN, OUT](
private val groupingKeys: Array[Int],
@transient private val returnType: TypeInformation[OUT])
extends RichMapFunction[IN, OUT] with ResultTypeQueryable[OUT] {
-
--- End diff --
yes,i will do it .
By the way Can we declare the style of the uniform Class / Object / Trait /
Method as shown in the following example?
1. Class/Object/Trait put each constructor argument on its own line,
indented four spaces:
If a class/object/trait extends anything, put each constructor argument on
its own line, indented four spaces and two spaces for extensions:
// wrong!
'''
class DataStreamAggregate(
window: LogicalWindow,
namedProperties: Seq[NamedWindowProperty],
cluster: RelOptCluster,
traitSet: RelTraitSet,
inputNode: RelNode,
namedAggregates: Seq[CalcitePair[AggregateCall, String]],
rowRelDataType: RelDataType,
inputType: RelDataType,
grouping: Array[Int])
extends SingleRel(cluster, traitSet, inputNode)
with FlinkAggregate
with DataStreamRel {
'''
// right!
'''
class DataStreamAggregate(
window: LogicalWindow,
namedProperties: Seq[NamedWindowProperty],
cluster: RelOptCluster,
traitSet: RelTraitSet,
inputNode: RelNode,
namedAggregates: Seq[CalcitePair[AggregateCall, String]],
rowRelDataType: RelDataType,
inputType: RelDataType,
grouping: Array[Int])
extends SingleRel(cluster, traitSet, inputNode)
with FlinkAggregate
with DataStreamRel {
'''
2. Methods with Numerous Arguments indented two spaces from the current
indent level:
// wrong!
'''private[flink] def createIncrementalAggregateReduceFunction(
aggregates: Array[Aggregate[_ <: Any]],
namedAggregates: Seq[CalcitePair[AggregateCall, String]],
inputType: RelDataType,
outputType: RelDataType,
groupings: Array[Int]): IncrementalAggregateReduceFunction = {
'''
// right!
'''
private[flink] def createIncrementalAggregateReduceFunction(
aggregates: Array[Aggregate[_ <: Any]],
namedAggregates: Seq[CalcitePair[AggregateCall, String]],
inputType: RelDataType,
outputType: RelDataType,
groupings: Array[Int]): IncrementalAggregateReduceFunction = {
'''
> Add incremental group window aggregation for streaming Table API
> ----------------------------------------------------------------
>
> Key: FLINK-4937
> URL: https://issues.apache.org/jira/browse/FLINK-4937
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Affects Versions: 1.2.0
> Reporter: Fabian Hueske
> Assignee: sunjincheng
>
> Group-window aggregates for streaming tables are currently not done in an
> incremental fashion. This means that the window collects all records and
> performs the aggregation when the window is closed instead of eagerly
> updating a partial aggregate for every added record. Since records are
> buffered, non-incremental aggregation requires more storage space than
> incremental aggregation.
> The DataStream API which is used under the hood of the streaming Table API
> features [incremental
> aggregation|https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/windows.html#windowfunction-with-incremental-aggregation]
> using a {{ReduceFunction}}.
> We should add support for incremental aggregation in group-windows.
> This is a follow-up task of FLINK-4691.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
