Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/3150#discussion_r96629837
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
---
@@ -226,27 +225,102 @@ object AggregateUtil {
aggregates,
groupingOffsetMapping,
aggOffsetMapping,
- intermediateRowArity,
+ intermediateRowArity + 1,// the addition one field is used to
store time attribute
outputType.getFieldCount)
+
+ case EventTimeSessionGroupWindow(_, _, gap) =>
+ val (startPos, endPos) = if (isTimeWindow(window)) {
+ computeWindowStartEndPropertyPos(properties)
+ } else {
+ (None, None)
+ }
+ new DataSetSessionWindowAggregateReduceGroupFunction(
+ aggregates,
+ groupingOffsetMapping,
+ aggOffsetMapping,
+ // the addition two fields are used to store window-start and
window-end attributes
+ intermediateRowArity + 2,
+ outputType.getFieldCount,
+ startPos,
+ endPos,
+ asLong(gap))
case _ =>
throw new UnsupportedOperationException(s"$window is currently not
supported on batch")
}
}
/**
+ * Create a
[[org.apache.flink.api.common.functions.GroupCombineFunction]] that
pre-aggregation
+ * for aggregates.
+ * The function returns intermediate aggregate values of all aggregate
function which are
+ * organized by the following format:
+ *
+ * {{{
+ * avg(x) aggOffsetInRow = 2 count(z)
aggOffsetInRow = 5
+ * | |
windowEnd(max(row-time)
+ * | |
|
+ * v v
v
+ *
+---------+---------+--------+--------+--------+--------+-----------+---------+
+ * |groupKey1|groupKey2| sum1 | count1 | sum2 | count2
|windowStart|windowEnd|
+ *
+---------+---------+--------+--------+--------+--------+-----------+---------+
+ * ^ ^
+ * | |
+ * sum(y) aggOffsetInRow = 4
windowStart(min(row-time))
+ *
+ * }}}
+ *
+ */
+
--- End diff --
Remove empty line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
