You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by dianfu <gi...@git.apache.org> on 2017/06/27 12:41:15 UTC
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
GitHub user dianfu opened a pull request:
https://github.com/apache/flink/pull/4195
[FLINK-7008] [cep] Update NFA state only when the NFA changes
Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful description of your changes.
- [ ] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the JIRA id)
- [ ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [ ] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis build has passed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/dianfu/flink FLINK-7008
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4195.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4195
----
commit 36b9baaaffecf8be32b5e8bb5887ccbd43430caf
Author: Dian Fu <fu...@alibaba-inc.com>
Date: 2017-06-27T12:38:53Z
[FLINK-7008] [cep] Update NFA state only when the NFA changes
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124499732
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -114,6 +115,28 @@ public DeweyNumber getVersion() {
return version;
}
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof ComputationState) {
+ ComputationState other = (ComputationState) obj;
+ return Objects.equals(state, other.state) &&
+ event == other.event &&
--- End diff --
For `ComputationState`, I think two `ComputationState` equals only if the last taken events of them point to the same event. So `.equals()` may be not correct. Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124512103
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +328,89 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 1858562682635302605L;
--- End diff --
OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124491894
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -152,6 +153,12 @@
private TypeSerializer<T> eventSerializer;
+ /**
+ * Flag indicating whether the matching status of the state machine has changed.
+ */
+ @VisibleForTesting
--- End diff --
I think we need setter anyway (see AbstractKeyedCEPPatternOperator:277). So I would make it private.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124494317
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
Currently for each input, we will deserialize the NFA and the `nfaChanged` field is `false` by default. So I think we don't need to do that. Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4195: [FLINK-7008] [cep] Update NFA state only when the NFA cha...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on the issue:
https://github.com/apache/flink/pull/4195
Hi @kl0u @dawidwys, could you help to review this PR when you are free? Thanks a lot.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124511647
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -237,8 +254,15 @@ public boolean isEmpty() {
computationState.getCounter());
newComputationStates = Collections.emptyList();
+ nfaChanged = true;
--- End diff --
Yes, you are right! I think then a test for that case would also be nice ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/4195
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124550121
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java ---
@@ -297,6 +297,85 @@ public void testKeyedAdvancingTimeWithoutElements() throws Exception {
}
@Test
+ public void testKeyedCEPOperatorNFAChanged() throws Exception {
--- End diff --
I meant rather checking if the resetting for the flag works correct. We could do it by veryfying correct number of invocations of `update` method. Also we should test it with both RocksDB and InMemory state backends.
We can do it with Mockito. Just a suggestion for the code:
@Test
public void testKeyedCEPOperatorNFAChanged() throws Exception {
String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend(new MemoryStateBackend());
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
KeyedCEPPatternOperator<Event, Integer> operator = new KeyedCEPPatternOperator<>(
Event.createTypeSerializer(),
true,
IntSerializer.INSTANCE,
new SimpleNFAFactory(),
true);
OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness = getCepTestHarness(operator);
try {
harness.setStateBackend(rocksDBStateBackend);
harness.open();
final ValueState nfaOperatorState = Whitebox.<ValueState>getInternalState(operator, "nfaOperatorState");
final ValueState nfaOperatorStateSpy = Mockito.spy(nfaOperatorState);
Whitebox.setInternalState(operator, "nfaOperatorState", nfaOperatorStateSpy);
Event startEvent = new Event(42, "c", 1.0);
SubEvent middleEvent = new SubEvent(42, "a", 1.0, 10.0);
Event endEvent = new Event(42, "b", 1.0);
harness.processElement(new StreamRecord<>(startEvent, 1L));
harness.processElement(new StreamRecord<>(new Event(42, "d", 1.0), 4L));
harness.processElement(new StreamRecord<Event>(middleEvent, 4L));
harness.processElement(new StreamRecord<>(endEvent, 4L));
Mockito.verify(nfaOperatorStateSpy, Mockito.times(3)).update(Mockito.any());
// get and verify the output
Queue<Object> result = harness.getOutput();
assertEquals(1, result.size());
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
} finally {
harness.close();
}
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124501193
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -114,6 +115,28 @@ public DeweyNumber getVersion() {
return version;
}
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof ComputationState) {
+ ComputationState other = (ComputationState) obj;
+ return Objects.equals(state, other.state) &&
+ event == other.event &&
--- End diff --
The combination of `Object.equals(event, other.event)` and `count == other.counter` tells exactly that. I would not base the implementation on reference equity.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124512134
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -152,6 +153,12 @@
private TypeSerializer<T> eventSerializer;
+ /**
+ * Flag indicating whether the matching status of the state machine has changed.
+ */
+ @VisibleForTesting
--- End diff --
OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124491446
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +328,89 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 1858562682635302605L;
--- End diff --
Anyway I feel it would be easier to read and understand if we extract a test for a `nfaChange` case at a time.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124697385
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java ---
@@ -297,6 +297,85 @@ public void testKeyedAdvancingTimeWithoutElements() throws Exception {
}
@Test
+ public void testKeyedCEPOperatorNFAChanged() throws Exception {
--- End diff --
Got it. Make sense and have updated the PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124500774
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -237,8 +254,15 @@ public boolean isEmpty() {
computationState.getCounter());
newComputationStates = Collections.emptyList();
+ nfaChanged = true;
--- End diff --
I think this may not always be true. For example, the `eventSharedBuffer.release`(NFA:250) may already have released the events that will be pruned. This will cause the pruning sharedBuffer do nothing. Thoughts?
Anyway, leave it here will be more safe.:)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124492599
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -237,8 +254,15 @@ public boolean isEmpty() {
computationState.getCounter());
newComputationStates = Collections.emptyList();
+ nfaChanged = true;
--- End diff --
Just a question/side comment. This case always happens in pair with pruning sharedBuffer right? (NFA:330)
I would still leave it here, cause it's better be safe than sorry.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124514053
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -237,8 +254,15 @@ public boolean isEmpty() {
computationState.getCounter());
newComputationStates = Collections.emptyList();
+ nfaChanged = true;
--- End diff --
OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124510489
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
Make sense.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124510810
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -114,6 +115,28 @@ public DeweyNumber getVersion() {
return version;
}
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof ComputationState) {
+ ComputationState other = (ComputationState) obj;
+ return Objects.equals(state, other.state) &&
+ event == other.event &&
--- End diff --
Make sense.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124492768
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -114,6 +115,28 @@ public DeweyNumber getVersion() {
return version;
}
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof ComputationState) {
+ ComputationState other = (ComputationState) obj;
+ return Objects.equals(state, other.state) &&
+ event == other.event &&
--- End diff --
how about comparing events with `.equals()`? It is not a primitive object.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4195: [FLINK-7008] [cep] Update NFA state only when the NFA cha...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on the issue:
https://github.com/apache/flink/pull/4195
@dawidwys @kl0u thanks a lot for the review comments, have updated the PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124455303
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +327,83 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 1858562682635302605L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).notFollowedBy("not").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = -6085237016591726715L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedByAny("middle").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("b");
+ }
+ }).oneOrMore().optional().allowCombinations().followedBy("middle2").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("d");
+ }
+ }).followedBy("end").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("e");
+ }
+ }).within(Time.milliseconds(10));
+
+ NFACompiler.NFAFactory<Event> nfaFactory = NFACompiler.compileFactory(pattern, Event.createTypeSerializer(), true);
+ NFA<Event> nfa = nfaFactory.createNFA();
+ nfa.process(new Event(1, "b", 1.0), 1L);
+ assertFalse(nfa.isNFAChanged());
+
+ nfa.nfaChanged = false;
--- End diff --
@dawidwys thanks a lot for the review. Have added more comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124481149
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
Shouldn't we reset nfaChanged to {{false}} here? Right now we do not update only if a Pattern did not start.
If we have a pattern like: A -> B, and sequence a, c, c, c, c, .... the NFA will be updated after each c nevertheless it does not change, won't it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124500395
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
This is only true for the `RocksDB` state backend. If you are using the `Heap` state backend, you do not (de)serialize at every access but you keep the same object. So the flag should be re-set.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124491313
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +328,89 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
--- End diff --
I feel the Pattern is a bit to complicated for this test. e.g. the notPattern does not introduce any new corner cases.
I miss a test for situation where only one CS is generated. You can simulate it with:
@Test
public void testNFAChangedOnOneNewComputationState() {
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
}).followedBy("a*").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 1858562682635302605L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
}).oneOrMore().optional().next("end").where(new IterativeCondition<Event>() {
private static final long serialVersionUID = 8061969839441121955L;
@Override
public boolean filter(Event value, Context<Event> ctx) throws Exception {
return value.getName().equals("b");
}
}).within(Time.milliseconds(10));
NFACompiler.NFAFactory<Event> nfaFactory = NFACompiler.compileFactory(pattern, Event.createTypeSerializer(), true);
NFA<Event> nfa = nfaFactory.createNFA();
// both the queue of ComputationStatus and eventSharedBuffer have changed
nfa.nfaChanged = false;
nfa.process(new Event(6, "start", 1.0), 6L);
nfa.nfaChanged = false;
nfa.process(new Event(6, "a", 1.0), 7L);
assertTrue("NFA status should change as the event matches the take condition of 'middle2' state", nfa.isNFAChanged());
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4195: [FLINK-7008] [cep] Update NFA state only when the NFA cha...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/4195
Thanks for the work @dianfu! It looks good now. Let's just wait for travis to give green on it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dianfu <gi...@git.apache.org>.
Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124512009
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +328,89 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
--- End diff --
Nice catch, will add this test to the test case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124491510
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
An IT test for it would be handy.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124496350
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java ---
@@ -270,10 +270,12 @@ private void updateLastSeenWatermark(long timestamp) {
}
private void updateNFA(NFA<IN> nfa) throws IOException {
- if (nfa.isEmpty()) {
- nfaOperatorState.clear();
- } else {
- nfaOperatorState.update(nfa);
+ if (nfa.isNFAChanged()) {
+ if (nfa.isEmpty()) {
+ nfaOperatorState.clear();
+ } else {
+ nfaOperatorState.update(nfa);
--- End diff --
Hmm, right we do access it from `StateBackend` each time. Not sure though if each time it is deserialized. I don't think there is such assumption. @kl0u could you shed some more light on it?
If I am right and it is not deserialized each time, I would reset the `nfaChanged` before writing it into the `StateBackend`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4195: [FLINK-7008] [cep] Update NFA state only when the ...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/4195#discussion_r124287999
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java ---
@@ -324,6 +327,83 @@ public boolean filter(Event value) throws Exception {
}
}
+ @Test
+ public void testNFAChange() {
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 1858562682635302605L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).notFollowedBy("not").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = -6085237016591726715L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedByAny("middle").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("b");
+ }
+ }).oneOrMore().optional().allowCombinations().followedBy("middle2").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("d");
+ }
+ }).followedBy("end").where(new IterativeCondition<Event>() {
+ private static final long serialVersionUID = 8061969839441121955L;
+
+ @Override
+ public boolean filter(Event value, Context<Event> ctx) throws Exception {
+ return value.getName().equals("e");
+ }
+ }).within(Time.milliseconds(10));
+
+ NFACompiler.NFAFactory<Event> nfaFactory = NFACompiler.compileFactory(pattern, Event.createTypeSerializer(), true);
+ NFA<Event> nfa = nfaFactory.createNFA();
+ nfa.process(new Event(1, "b", 1.0), 1L);
+ assertFalse(nfa.isNFAChanged());
+
+ nfa.nfaChanged = false;
--- End diff --
Could you add comments for each assertion, what situation in NFA it covers?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---