Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/6171#discussion_r198474417
--- Diff:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -330,77 +328,85 @@ private boolean isStateTimedOut(final
ComputationState state, final long timesta
}
}
- discardComputationStatesAccordingToStrategy(
- sharedBuffer, computationStates, result,
afterMatchSkipStrategy);
+ if (!potentialMatches.isEmpty()) {
+ nfaState.setStateChanged();
+ }
+
+ List<Map<String, List<T>>> result = new ArrayList<>();
+ if (afterMatchSkipStrategy.isSkipStrategy()) {
+ processMatchesAccordingToSkipStrategy(sharedBuffer,
+ nfaState,
+ afterMatchSkipStrategy,
+ potentialMatches,
+ result);
+ } else {
+ for (ComputationState match : potentialMatches) {
+
result.add(sharedBuffer.materializeMatch(sharedBuffer.extractPatterns(match.getPreviousBufferEntry(),
--- End diff --
Instead of accessing the state for every match, why not passing all the
matches to the shared buffer, and try to fetch the common ones only once. If 2
matches A and B share event with id = 2, we fetch from state only once.
---
