SEPURI-SAI-KRISHNA commented on code in PR #29410:
URL: https://github.com/apache/flink/pull/29410#discussion_r4244579828
##########
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);
Review Comment:
I tried it: copy only inside setAccumulatorOfPrevRow and
setAccumulatorOfPrevId, plain get elsewhere. Green, 26/26 in
NonTimeOverAggregateITCase and 54/54 in the harness tests.
That only says the suite does not catch it. Those other reads are safe
because each one is handed to setAccumulators and the handler is pointed at
another accumulator before anything accumulates into it. Nothing enforces that
ordering, and the earlier version of this patch already shipped a hole of
exactly that kind by gating on the accumulator type and missing BITMAP. So I
would rather keep the copy at the state boundary. Happy to narrow it if you
prefer.
##########
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:
The write copy has to stay. Dropping it on its own fails 8 of 26 in
NonTimeOverAggregateITCase, all on the HEAP run: LAG, ARRAY_AGG,
BITMAP_BUILD_CARDINALITY, COLLECT and COUNT(DISTINCT). The harness tests stay
54/54, so the ITCase is the only thing that catches it.
An allowlist on the aggregate would not be safe. COUNT(DISTINCT v) is one of
those 8 and its accumulator carries a MapView, so seeing COUNT tells you
nothing about the fields. Checking every field serializer for isImmutableType
would be sound, MapViewSerializer and BitmapSerializer both return false, but I
do not think it earns the extra branch here.
Every record already reads the whole sorted list back through
sortedListState.value() and scans it linearly in findIndexOfSortKey. An append
copies two accumulators, and an insert in between copies two more per following
row, each paired with a state get or put that costs more than the copy itself.
If we want to cut work on this path, the state round trips are where it is, not
the copy.
--
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]
