You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2017/05/17 12:38:55 UTC
[9/9] flink git commit: [FLINK-6609] [cep] Fix wrong version
assignment with multiple TAKEs.
[FLINK-6609] [cep] Fix wrong version assignment with multiple TAKEs.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/00ce3f1b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/00ce3f1b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/00ce3f1b
Branch: refs/heads/master
Commit: 00ce3f1b12c7d7bf996d5f91bf006f0e18a719e7
Parents: 7a54d05
Author: Dawid Wysakowicz <da...@getindata.com>
Authored: Wed May 17 09:16:08 2017 +0200
Committer: kkloudas <kk...@gmail.com>
Committed: Wed May 17 14:37:35 2017 +0200
----------------------------------------------------------------------
.../main/java/org/apache/flink/cep/nfa/NFA.java | 11 ++-
.../org/apache/flink/cep/nfa/NFAITCase.java | 82 ++++++++++++++++++++
2 files changed, 89 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/00ce3f1b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
index ab5cd8e..a977a7f 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
@@ -418,6 +418,7 @@ public class NFA<T> implements Serializable {
final List<StateTransition<T>> edges = outgoingEdges.getEdges();
int takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
int ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ int totalTakeToSkip = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
final List<ComputationState<T>> resultingComputationStates = new ArrayList<>();
for (StateTransition<T> edge : edges) {
@@ -433,7 +434,9 @@ public class NFA<T> implements Serializable {
version = computationState.getVersion().increase(toIncrease);
} else {
//IGNORE after PROCEED
- version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ version = computationState.getVersion()
+ .increase(totalTakeToSkip + ignoreBranchesToVisit)
+ .addStage();
ignoreBranchesToVisit--;
}
@@ -457,8 +460,8 @@ public class NFA<T> implements Serializable {
final T previousEvent = computationState.getEvent();
- final DeweyNumber currentVersion = computationState.getVersion();
- final DeweyNumber nextVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ final DeweyNumber currentVersion = computationState.getVersion().increase(takeBranchesToVisit);
+ final DeweyNumber nextVersion = new DeweyNumber(currentVersion).addStage();
takeBranchesToVisit--;
final int counter;
@@ -573,7 +576,7 @@ public class NFA<T> implements Serializable {
}
private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
- return takeBranches == 0 && ignoreBranches == 0 ? 0 : ignoreBranches + 1;
+ return takeBranches == 0 && ignoreBranches == 0 ? 0 : ignoreBranches + Math.max(1, takeBranches);
}
private OutgoingEdges<T> createDecisionGraph(ComputationState<T> computationState, T event) {
http://git-wip-us.apache.org/repos/asf/flink/blob/00ce3f1b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
index 012e112..d00bbb7 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
@@ -3974,6 +3974,88 @@ public class NFAITCase extends TestLogger {
}
@Test
+ public void testMultipleTakesVersionCollision() {
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+
+ Event startEvent = new Event(40, "c", 1.0);
+ Event middleEvent1 = new Event(41, "a", 2.0);
+ Event middleEvent2 = new Event(41, "a", 3.0);
+ Event middleEvent3 = new Event(41, "a", 4.0);
+ Event middleEvent4 = new Event(41, "a", 5.0);
+ Event middleEvent5 = new Event(41, "a", 6.0);
+ Event end = new Event(44, "b", 5.0);
+
+ inputEvents.add(new StreamRecord<>(startEvent, 1));
+ inputEvents.add(new StreamRecord<>(middleEvent1, 3));
+ inputEvents.add(new StreamRecord<>(middleEvent2, 4));
+ inputEvents.add(new StreamRecord<>(middleEvent3, 5));
+ inputEvents.add(new StreamRecord<>(middleEvent4, 6));
+ inputEvents.add(new StreamRecord<>(middleEvent5, 7));
+ inputEvents.add(new StreamRecord<>(end, 10));
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedBy("middle1").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).oneOrMore().allowCombinations().followedBy("middle2").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).oneOrMore().allowCombinations().followedBy("end").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("b");
+ }
+ });
+
+ NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
+
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
+
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
+
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent4, middleEvent5, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, end),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent5, end),
+
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end)
+ ));
+ }
+
+ @Test
public void testZeroOrMoreSameElement() {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();