SEPURI-SAI-KRISHNA commented on code in PR #29410:
URL: https://github.com/apache/flink/pull/29410#discussion_r4217538776
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/AbstractNonTimeUnboundedPrecedingOver.java:
##########
@@ -195,6 +199,10 @@ public void open(OpenContext openContext) throws Exception
{
sortKeyEqualiser =
generatedSortKeyEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ // Initialize accumulator equaliser
+ accEqualiser =
+
generatedAccEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
Review Comment:
Done, and this is better than what I had. Every get and put of accMapState
now goes through getAccFromState and putAccInState, which copy with the
accumulator serializer every time, for every accumulator. The remove and the
TTL cleanupState still use the state directly, neither of them hands an
accumulator out.
My version gated the copy on the accumulator holding a RAW field, which was
too narrow. BITMAP is
its own LogicalTypeRoot, so BITMAP_BUILD_CARDINALITY_AGG was never copied,
and on the same data
heap returned {10=1, 20=3, 25=4, 30=2} against {10=1, 20=2, 25=3, 30=4} on
RocksDB. Added
testBitmapBuildCardinalityInsertsInBetween.
The copy on put is also what the reset at the end of processElement needed,
so I dropped the
conditional copy I had on FLINK-40737.
Red/green on this branch: 4 of the 12 runs fail without the copy, all of
them on heap, and all 12
pass with it. The four are the tests named above.
One trade off worth naming: the copy is unconditional, so it also runs on
paths that only read an
accumulator, and on RocksDB where the backend already hands back its own
copy. I kept it uniform
because narrowing it is what produced the bug I just fixed. Happy to
restrict it if you would
rather not pay that on the per record path.
--
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]
