[FLINK-7043] Optimize state handling in NFA

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.3.1
• Component/s: Library / CEP
• Assignee: Dawid Wysakowicz
• Reporter: Kostas Kloudas
• Created: 29/Jun/17 17:03
• Updated: 05/Jun/18 13:46
• Resolved: 05/Jun/18 13:46

Description

The checkpointed state of the NFA is consisted of:
1) The States/ windowTime/ handleTimeout flag (also contains the IterativeConditions)
2) The sharedBuffer
3) The computationStates

From these 3:
1) is static, as it is the description of the NFA which is static throughout the lifetime of the NFA, and
2) and 3) that are dynamic, as at any point they depend on the already consumed input.

This issue proposes to leverage the fact the 1 is static and store it on a separate keyed valueState than the NFA. This way, in the AbstractKeyedCEPPatternOperator.getNFA() which is called at each watermark and for each active key, we will just serialize/deserialize the dynamic state, while the static one we will serialize it only upon checkpointing, and deserialize it upon restoring.

We need to checkpoint also the static state because we want in the future to be able to add Patterns at runtime to the same operator, e.g. through a coStream.

So given that now we must have a way to match the dynamic with the static state of the NFA, we will have to introduce a name (or id) for a given Pattern/NFA. This will be stored with both the
static and the dynamic state, and upon restoring we will be able to match the two based on that name. This name will be user-defined and can be specified through a `name()` method in the Pattern class (like the begin()).

Issue Links

• duplicates: FLINK-8725 Separate NFA-state from NFA in CEP library