bvarghese1 commented on code in PR #29410:
URL: https://github.com/apache/flink/pull/29410#discussion_r4235496199
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/AbstractNonTimeUnboundedPrecedingOver.java:
##########
@@ -422,6 +436,36 @@ RowData setAccumulatorAndGetValue(RowData accumulator)
throws Exception {
return aggFuncs.getValue();
}
+ /**
+ * Reads the accumulator stored for a sort key, as a copy the caller owns.
+ *
+ * <p>An accumulator can hold mutable content, either a data view or a
field such as a bitmap,
+ * an array or a {@code byte[]}. The aggregate functions change that
content in place, and the
+ * heap state backend hands back the very object it stores. Without a
copy, accumulating into
+ * one sort key's accumulator would also change the one state holds for
another.
+ *
+ * @param accKey the sort key to read the accumulator of
+ * @return a private copy of the stored accumulator, or null if there is
none
+ */
+ RowData getAccFromState(RowData accKey) throws Exception {
+ final RowData acc = accMapState.get(accKey);
+ return acc == null ? null : accSerializer.copy(acc);
+ }
+
+ /**
+ * Stores the accumulator of a sort key, as a copy state owns.
+ *
+ * <p>The aggregate functions keep changing their accumulator after it has
been stored, and the
+ * reset at the end of {@link #processElement} clears its data views. The
heap state backend
+ * keeps the object it was given, so the copy is what keeps the stored
accumulator intact.
+ *
+ * @param accKey the sort key to store the accumulator of
+ * @param acc the accumulator to store
+ */
+ void putAccInState(RowData accKey, RowData acc) throws Exception {
+ accMapState.put(accKey, accSerializer.copy(acc));
Review Comment:
This is also on the hot path. Can we skip the copy when it doesnt matter? If
every accumulator field is immutable, the generated handler never changes the
stored row in place. This can be done once in the `open()` and we can skip for
the most common cases ie. `COUNT/SUM/MIN/MAX/AVG/ROW_NUMBER/RANK`.
If its indeed cheap like @MartijnVisser suggested I would avoid this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
