[
https://issues.apache.org/jira/browse/FLINK-30208?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17644784#comment-17644784
]
xljtswf commented on FLINK-30208:
---------------------------------
[~kevin.cyj] Yes, this is the most cases.
Indeed, there are 2 conditions we can improve:
1. when maxCount==1, i.e. we will trigger on every element. Thus we even do not
need to use the state at all, there will be no get and update. And stateDesc do
not need to be instantiated which can save the serialized memory. But this will
change the serialized bytes of CountTrigger, I do not know whether it will
change the back-compatibilaty.
2. when maxCount > 1. For the last element, we can save 1 add operation.
> avoid unconditional state update in CountTrigger#onElement
> ----------------------------------------------------------
>
> Key: FLINK-30208
> URL: https://issues.apache.org/jira/browse/FLINK-30208
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream
> Reporter: xljtswf
> Priority: Major
>
> In current CountTrigger#onElement, when one element is received, the state is
> updated unconditionally, and we then fetch the state again to check whether
> we need to clear the state. This implies we may update the state 2 times to
> process one element. I suppose to make following simplification:
> public TriggerResult onElement(Object element, long timestamp, W window,
> TriggerContext ctx)
> throws Exception {
> TriggerResult triggerResult;
> if (maxCount > 1) {
> ReducingState<Long> countState =
> ctx.getPartitionedState(stateDesc);
> Long currentCount = countState.get();
> if (currentCount == null || currentCount < maxCount - 1) {
> countState.add(1L);
> triggerResult = TriggerResult.CONTINUE;
> } else {
> countState.clear();
> triggerResult = TriggerResult.FIRE;
> }
> } else {
> triggerResult = TriggerResult.FIRE;
> }
> return triggerResult;
> }
> If this is approved, I will make a pr then.
> Thanks!
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
