You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by dawidwys <gi...@git.apache.org> on 2017/03/06 10:29:04 UTC
[GitHub] flink pull request #3477: [Flink3318] Add support for quantifiers to CEP's p...
GitHub user dawidwys opened a pull request:
https://github.com/apache/flink/pull/3477
[Flink3318] Add support for quantifiers to CEP's pattern API
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/dawidwys/flink flink3318
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3477.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 #3477
----
commit 89840a08cb29384f2ce196cb45d44015f5b79f34
Author: Dawid Wysakowicz <da...@getindata.com>
Date: 2017-03-06T09:57:36Z
[FLINK-3318] Add support for quantifiers to CEP's pattern API
commit 660896f8aecfd96d7b6c1e4e548fea1d6811b9e7
Author: Dawid Wysakowicz <da...@getindata.com>
Date: 2017-03-06T10:27:29Z
[FLINK-3318] Added docs
----
---
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 #3477: [Flink-3318] Add support for quantifiers to CEP's pattern...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
Thanks @kl0u for the comment. I was wondering if it would be possible(and beneficial) to restrict it on the interface level and come up with a class hierachy like :
```java
interface IPattern {
/**
Getters for properties e.g.
*/
String getName()
Quantifier getQuantifier()
}
class Pattern implements IPattern {
public Pattern<T, F> where(FilterFunction<F> newFilterFunction)
public Pattern<T, F> where(FilterFunction<F> newFilterFunction)
public QuantifiedPattern<T, F> zeroOrMore(final boolean eager)
...
}
class QuantifiedPattern implements IPattern {
private Pattern<T, F> wrappedPattern;
public QuantifiedPattern<T, F> where(FilterFunction<F> newFilterFunction)
public String getName() {
return wrappedPattern.getName();
}
public String getQuantifier() {
return wrappedPattern.getQuantifier();
}
/* no quantifier specific methods */
}
```
You can see a working example for java-cep in my fork: https://github.com/dawidwys/flink/tree/cepQuantifierAPI
Not sure if the hierarchy is not too complex, and simple checks with exceptions would be better.
We would also need to duplicate this hierarchy in the scala part.
@kl0u What do you think?
---
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 #3477: [Flink-3318] Add support for quantifiers to CEP's pattern...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Hi @dawidwys , I will do this in the beginning of next week at the latest.
One thing that I saw is that in the `Pattern` class you do not check for conflicting configurations e.g. oneToMany and zeroToMany. It would be nice to have that with a good error message.
---
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 #3477: [Flink-3318] Add support for quantifiers to CEP's pattern...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
@kl0u could you have a look at this 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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106163326
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
--- End diff --
Ad.3 There is one problem with the atMost. Consider atMost(2).optional().where(a). It will be translated into states:
`q1? q2?` . For a sequence `a1 a2`. It will generate matches `a1; a1; a2; a2; a1 a2`. Notice duplicated `a1` and `a2`. It is because a single event can match either on `q1` or at `q2`.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105697591
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -52,14 +57,35 @@ public ComputationState(
this.timestamp = timestamp;
this.version = version;
this.startTimestamp = startTimestamp;
+ this.previousState = previousState;
+ }
+
--- End diff --
Move the 3 constructors at the end of the class.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106142891
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
--- End diff --
I mean the check is performed outside of the method anyway, so we would perform it twice. I think the name(which I hope implies it already) is enough.
If we were to be defensive here I think the best check would be:
`Preconditions.checkState(currentPattern.getQuantifier().isAtLeastOne(),"There is a bug in implementation. Please file a JIRA);` What do you think?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105856015
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java ---
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.cep.pattern;
+
+public enum Quantifier {
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105700201
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
--- End diff --
Move both the arguments to different lines.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106109112
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -74,11 +80,35 @@ public long getStartTimestamp() {
return state;
}
+ public State<T> getPreviousState() {
+ return previousState;
+ }
+
public T getEvent() {
return event;
}
public DeweyNumber getVersion() {
return version;
}
+
+ public static <T> ComputationState<T> createStartState(final State<T> state) {
+ Preconditions.checkArgument(state.isStart());
+ return new ComputationState<>(state, null, null, -1L, new DeweyNumber(1), -1L);
+ }
+
+ public static <T> ComputationState<T> createStartState(final State<T> state, final DeweyNumber version) {
+ Preconditions.checkArgument(state.isStart());
+ return new ComputationState<>(state, null, null, -1L, version, -1L);
+ }
+
--- End diff --
Indent the arguments to separate them from the body of the method. This will make the code easier to read.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Hi @eliaslevy , @dawidwys is right, in fact if the state's name is "a" and it has 2 matching events, there will be two returned keys "a_0" and "a_1". This is definitely counterintuitive and the reason it is still there is for backwards compatibility with previous versions.
I think this should be addressed but I would also suggest to also start a discussion in the mailing list to see if people are ok with this. Unfortunately we are not aware if anyone is using the `CEP` library at the moment.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106108080
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
--- End diff --
You can add a check:
`Preconditions.checkArgument(currentPattern.getQuantifier().isAtLeastOne());`
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105704399
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
+ 5.0;
}
})
- .followedBy("middle-second").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ .followedBy("middle-second").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getName().equals("next-one");
+ return value.getName()
+ .equals(
+ "next-one");
}
})
.followedBy("end").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
public boolean filter(Event value) throws Exception {
- return value.getName().equals("end");
+ return value.getName()
--- End diff --
this can be as before, in one line.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106107383
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
--- End diff --
From what I understand, the `ignoredState` is needed only in the case where we have a `FollowedByPattern`. In this case the code could become:
```
private void convertToSingletonState(
final State<T> sinkState,
final State<T> sourceState) {
final FilterFunction<T> currentFilterFunction =
(FilterFunction<T>) currentPattern.getFilterFunction();
final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
sourceState.addTake(sinkState, currentFilterFunction);
if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
sourceState.addProceed(sinkState, trueFunction);
}
if (currentPattern instanceof FollowedByPattern) {
final State<T> ignoreState;
if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
ignoreState.addTake(sinkState, currentFilterFunction);
states.add(ignoreState);
} else {
ignoreState = sourceState;
}
sourceState.addIgnore(ignoreState, trueFunction);
}
}
```
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105698626
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -209,33 +204,66 @@ public boolean equals(Object obj) {
return nonDuplicatingTypeSerializer.equals(other.nonDuplicatingTypeSerializer) &&
sharedBuffer.equals(other.sharedBuffer) &&
states.equals(other.states) &&
- windowTime == other.windowTime &&
- startEventCounter == other.startEventCounter;
+ windowTime == other.windowTime;
} else {
return false;
}
}
@Override
public int hashCode() {
- return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime, startEventCounter);
+ return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime);
+ }
+
--- End diff --
Why is this method is needed? Is the name comparison enough? And if it is needed indeed, I think it should move to the `State` class with a comment on why it is needed.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105699655
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
--- End diff --
Unbox these two `Integer`s to `int`s.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106106765
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
--- End diff --
You do not need to declare it here and set it 3 lines below, you can just create the variable later when you set it:
`State<T> sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);`
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105851817
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -71,19 +70,16 @@
private final NonDuplicatingTypeSerializer<T> nonDuplicatingTypeSerializer;
// Buffer used to store the matched events
--- End diff --
I've changed it to String because, one `Pattern` can be compiled into multiple `State`s. I assumed the name is unique for `Pattern` and when the resulting matches are returned we want to have events grouped by the `Pattern` name in `Map<String, Event>`.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105853157
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -71,19 +70,16 @@
private final NonDuplicatingTypeSerializer<T> nonDuplicatingTypeSerializer;
// Buffer used to store the matched events
- private final SharedBuffer<State<T>, T> sharedBuffer;
+ private final SharedBuffer<String, T> sharedBuffer;
// Set of all NFA states
- private final Set<State<T>> states;
--- End diff --
Reverted. It was needed as for a while I had other `State::equals` semantics.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854967
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
+ return 0;
+ }
+
+ return ignoreBranches + 1;
+ }
+
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106162103
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final State<T> firstState = new State<>(currentPattern.getName(), stateType);
+
+ firstState.addTake(sinkState, currentFilterFunction);
+ if (currentPattern instanceof FollowedByPattern) {
+ if (currentPattern.getQuantifier() == Quantifier.ONE_OR_MORE_COMBINATIONS) {
--- End diff --
I changed the `Quantifier`. How do you like it now?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105698150
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -71,19 +70,16 @@
private final NonDuplicatingTypeSerializer<T> nonDuplicatingTypeSerializer;
// Buffer used to store the matched events
- private final SharedBuffer<State<T>, T> sharedBuffer;
+ private final SharedBuffer<String, T> sharedBuffer;
// Set of all NFA states
- private final Set<State<T>> states;
--- End diff --
Why `List` and not `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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854241
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -209,33 +204,66 @@ public boolean equals(Object obj) {
return nonDuplicatingTypeSerializer.equals(other.nonDuplicatingTypeSerializer) &&
sharedBuffer.equals(other.sharedBuffer) &&
states.equals(other.states) &&
- windowTime == other.windowTime &&
- startEventCounter == other.startEventCounter;
+ windowTime == other.windowTime;
} else {
return false;
}
}
@Override
public int hashCode() {
- return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime, startEventCounter);
+ return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime);
+ }
+
+ private static <T> boolean isEquivalentState(final State<T> s1, final State<T> s2) {
+ return s1.getName().equals(s2.getName());
}
/**
- * Comparator used for imposing the assumption that IGNORE is always the last StateTransition in a state.
- */
- private interface StateTransitionComparator<T> extends Comparator<StateTransition<T>>, Serializable {}
- private final Comparator<StateTransition<T>> stateTransitionComparator = new StateTransitionComparator<T>() {
- private static final long serialVersionUID = -2775474935413622278L;
+ * Structures to keep decisions based on the transition actions, that counts the number of taken actions.
+ */
+ private static class OutgoingEdges<T> {
+ private List<StateTransition<T>> edges = new ArrayList<>();
- @Override
- public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
- if (o1.getAction() == o2.getAction()) {
- return 0;
+ private final State<T> currentState;
+
+ private int totalTakeBranches = 0;
+ private int totalIgnoreBranches = 0;
+
+ OutgoingEdges(final State<T> currentState) {
+ this.currentState = currentState;
+ }
+
+ void add(StateTransition<T> edge) {
+
+ if (!isSelfIgnore(edge)) {
+ if (edge.getAction() == StateTransitionAction.IGNORE) {
+ totalIgnoreBranches++;
+ } else if (edge.getAction() == StateTransitionAction.TAKE) {
+ totalTakeBranches++;
+ }
}
- return o1.getAction() == StateTransitionAction.IGNORE ? 1 : -1;
+
+ edges.add(edge);
+ }
+
+ int getTotalIgnoreBranches() {
+ return totalIgnoreBranches;
+ }
+ int getTotalTakeBranches() {
+ return totalTakeBranches;
}
- };
+
+ List<StateTransition<T>> getEdges() {
+ return edges;
+ }
+
--- End diff --
No as explained before they can be different states in case of transitions.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105857530
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java ---
@@ -18,9 +18,13 @@
package org.apache.flink.cep.nfa.compiler;
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105857595
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
+ 5.0;
}
})
- .followedBy("middle-second").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ .followedBy("middle-second").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getName().equals("next-one");
+ return value.getName()
+ .equals(
+ "next-one");
}
})
.followedBy("end").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
public boolean filter(Event value) throws Exception {
- return value.getName().equals("end");
+ return value.getName()
--- End diff --
Reverted
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105704267
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
--- End diff --
this can be as before, in one line. In general we try to minimize unnecessary 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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854765
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Hi @dawidwys , I will start reviewing your PR now.
I would say that for now it makes sense to have it with exceptions and not change the `Pattern` hierarchy. The reason is:
1) not sure what the above split will imply to backwards compatibility
2) there are some additional features that will be nice to have, and they all affect also how patterns are defined. A refactoring of the `Pattern` hierarchy would make more sense as soon as we have everything in place, and we can factor in this knowledge when designing.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105851882
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -41,8 +43,11 @@
// Timestamp of the first element in the pattern
private final long startTimestamp;
- public ComputationState(
+ private final State<T> previousState;
+
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
Great! Thanks for your help @kl0u !
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854471
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105855625
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java ---
@@ -217,23 +217,22 @@ public void prune(long pruningTimestamp) {
// use a depth first search to reconstruct the previous relations
while (!extractionStates.isEmpty()) {
ExtractionState<K, V> extractionState = extractionStates.pop();
- DeweyNumber currentVersion = extractionState.getVersion();
// current path of the depth first search
Stack<SharedBufferEntry<K, V>> currentPath = extractionState.getPath();
// termination criterion
- if (currentVersion.length() == 1) {
- LinkedHashMultimap<K, V> completePath = LinkedHashMultimap.create();
+ if (extractionState.getEntry() == null) {
+ final LinkedHashMultimap<K, V> completePath = LinkedHashMultimap.create();
while(!currentPath.isEmpty()) {
- SharedBufferEntry<K, V> currentEntry = currentPath.pop();
+ final SharedBufferEntry<K, V> currentEntry = currentPath.pop();
completePath.put(currentEntry.getKey(), currentEntry.getValueTime().getValue());
}
result.add(completePath);
} else {
- SharedBufferEntry<K, V> currentEntry = extractionState.getEntry();
--- End diff --
If you mean outside the `if-else` then right. I've done that.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Hi @dawidwys , the following produces no output, which should not be the case. The same holds if you change the pattern to `oneOrMore`.
```
public void testZeroOrMore() {
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(42, "a", 3.0);
Event middleEvent3 = new Event(43, "a", 4.0);
Event end1 = new Event(44, "b", 5.0);
Event end2 = new Event(45, "d", 6.0);
Event end3 = new Event(46, "d", 7.0);
Event end4 = new Event(47, "e", 8.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<>(end1, 6));
inputEvents.add(new StreamRecord<>(end2, 7));
inputEvents.add(new StreamRecord<>(end3, 8));
inputEvents.add(new StreamRecord<>(end4, 9));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
}).zeroOrMore();
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
Set<Set<Event>> resultingPatterns = new HashSet<>();
List<Collection<Event>> allPatterns = new ArrayList<>();
for (StreamRecord<Event> inputEvent : inputEvents) {
Collection<Map<String, Event>> patterns = nfa.process(
inputEvent.getValue(),
inputEvent.getTimestamp()).f0;
for (Map<String, Event> foundPattern : patterns) {
System.out.println(foundPattern);
}
}
}
```
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105697452
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -41,8 +43,11 @@
// Timestamp of the first element in the pattern
private final long startTimestamp;
- public ComputationState(
+ private final State<T> previousState;
+
--- End diff --
Insert one tab before each argument so that it is easier to separate the arguments from the body of the method.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106108841
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final State<T> firstState = new State<>(currentPattern.getName(), stateType);
+
+ firstState.addTake(sinkState, currentFilterFunction);
+ if (currentPattern instanceof FollowedByPattern) {
+ if (currentPattern.getQuantifier() == Quantifier.ONE_OR_MORE_COMBINATIONS) {
+ firstState.addIgnore(FilterFunctions.<T>trueFunction());
+ } else {
+ firstState.addIgnore(new NotFilterFunction<>(currentFilterFunction));
+ }
}
+ return firstState;
+ }
- beginningState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- currentState,
- (FilterFunction<T>) currentPattern.getFilterFunction()
- ));
+ @SuppressWarnings("unchecked")
+ private void convertToLooping(
+ final State<T> sinkState,
+ final State<T> sourceState) {
- return new NFAFactoryImpl<T>(inputTypeSerializer, windowTime, new HashSet<>(states.values()), timeoutHandling);
+ final FilterFunction<T> filterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.<T>trueFunction();
+
+ sourceState.addProceed(sinkState, trueFunction);
+ sourceState.addTake(filterFunction);
+ if (currentPattern instanceof FollowedByPattern) {
+ final State<T> ignoreState = new State<>(
+ currentPattern.getName(),
+ State.StateType.Normal);
+
+
+ final FilterFunction<T> ignoreCondition;
+ if (currentPattern.getQuantifier() == Quantifier.ZERO_OR_MORE_COMBINATIONS ||
--- End diff --
If you add the flag I proposed above in the `Quantifier` class, this check will become clearer I think (`currentPattern.getQuantifier().isCombinations()`) or something like this.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105704530
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
--- End diff --
this can be as before, in one line.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105700500
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
+ return 0;
+ }
+
+ return ignoreBranches + 1;
+ }
+
--- End diff --
Move arguments in the same line
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105702140
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/InvalidPatternException.java ---
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.cep.pattern;
+
+/**
+ * Thrown when the given {@link Pattern} was constructed illegaly e.g when
+ * more than quantifier provided.
+ */
+class InvalidPatternException extends RuntimeException {
--- End diff --
Instead of creating a new `Exception` I suggest we should use the existing `MalformedPatternException`. You can move it to this package, as it is used more broadly now.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105851855
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -52,14 +57,35 @@ public ComputationState(
this.timestamp = timestamp;
this.version = version;
this.startTimestamp = startTimestamp;
+ this.previousState = previousState;
+ }
+
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105702715
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java ---
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.cep.pattern;
+
+public enum Quantifier {
--- End diff --
Here you can add some flags, like `isLooping`, `isAtLeastOnce`, etc and set them accordingly in the constructors. This will allow you to move all the `is*` methods like `isLooping()`, `isAtLeastOnce` from the `NFACompiler` here.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105701064
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java ---
@@ -217,23 +217,22 @@ public void prune(long pruningTimestamp) {
// use a depth first search to reconstruct the previous relations
while (!extractionStates.isEmpty()) {
ExtractionState<K, V> extractionState = extractionStates.pop();
- DeweyNumber currentVersion = extractionState.getVersion();
// current path of the depth first search
Stack<SharedBufferEntry<K, V>> currentPath = extractionState.getPath();
// termination criterion
- if (currentVersion.length() == 1) {
- LinkedHashMultimap<K, V> completePath = LinkedHashMultimap.create();
+ if (extractionState.getEntry() == null) {
+ final LinkedHashMultimap<K, V> completePath = LinkedHashMultimap.create();
while(!currentPath.isEmpty()) {
- SharedBufferEntry<K, V> currentEntry = currentPath.pop();
+ final SharedBufferEntry<K, V> currentEntry = currentPath.pop();
completePath.put(currentEntry.getKey(), currentEntry.getValueTime().getValue());
}
result.add(completePath);
} else {
- SharedBufferEntry<K, V> currentEntry = extractionState.getEntry();
--- End diff --
The `SharedBufferEntry<K, V> currentEntry = extractionState.getEntry();` can be moved outside the for-loop right?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105702323
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java ---
@@ -18,9 +18,13 @@
package org.apache.flink.cep.nfa.compiler;
--- End diff --
Unused imports.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Merged this! Thanks for the work @dawidwys.
Can you close this PR and the JIRA?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105701669
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/StateTransition.java ---
@@ -55,6 +65,7 @@ public boolean equals(Object obj) {
StateTransition<T> other = (StateTransition<T>) obj;
--- End diff --
You have updated the `equals`, you should also update the `hashCode` and the `toString` accordingly.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105704335
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
+ 5.0;
}
})
- .followedBy("middle-second").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ .followedBy("middle-second").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getName().equals("next-one");
+ return value.getName()
+ .equals(
+ "next-one");
}
})
.followedBy("end").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
public boolean filter(Event value) throws Exception {
- return value.getName().equals("end");
+ return value.getName()
+ .equals("end");
}
});
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
List<Map<String, Event>> resultingPatterns = new ArrayList<>();
- for (StreamRecord<Event> inputEvent: inputEvents) {
+ for (StreamRecord<Event> inputEvent : inputEvents) {
--- End diff --
this can be as before, in one line.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105857618
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
--- End diff --
Reverted
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105855949
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/InvalidPatternException.java ---
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.cep.pattern;
+
+/**
+ * Thrown when the given {@link Pattern} was constructed illegaly e.g when
+ * more than quantifier provided.
+ */
+class InvalidPatternException extends RuntimeException {
--- End diff --
Done.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106108553
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final State<T> firstState = new State<>(currentPattern.getName(), stateType);
+
+ firstState.addTake(sinkState, currentFilterFunction);
+ if (currentPattern instanceof FollowedByPattern) {
+ if (currentPattern.getQuantifier() == Quantifier.ONE_OR_MORE_COMBINATIONS) {
+ firstState.addIgnore(FilterFunctions.<T>trueFunction());
+ } else {
+ firstState.addIgnore(new NotFilterFunction<>(currentFilterFunction));
+ }
}
+ return firstState;
+ }
- beginningState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- currentState,
- (FilterFunction<T>) currentPattern.getFilterFunction()
- ));
+ @SuppressWarnings("unchecked")
+ private void convertToLooping(
+ final State<T> sinkState,
+ final State<T> sourceState) {
- return new NFAFactoryImpl<T>(inputTypeSerializer, windowTime, new HashSet<>(states.values()), timeoutHandling);
+ final FilterFunction<T> filterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
--- End diff --
You can add a check: `Preconditions.checkArgument(currentPattern.getQuantifier().isLooping());`
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106108473
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final State<T> firstState = new State<>(currentPattern.getName(), stateType);
+
+ firstState.addTake(sinkState, currentFilterFunction);
+ if (currentPattern instanceof FollowedByPattern) {
+ if (currentPattern.getQuantifier() == Quantifier.ONE_OR_MORE_COMBINATIONS) {
--- End diff --
In the `Quantifier` you can add a flag like `isCombinations` for the `ONE_OR_MORE_COMBINATIONS` and the `ZERO_OR_MORE_COMBINATIONS` and check it here. Given that you already check for `atLeastOne` in the check I proposed above, this will be sufficient.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by eliaslevy <gi...@git.apache.org>.
Github user eliaslevy commented on the issue:
https://github.com/apache/flink/pull/3477
Am I missing something or is there no way to get access to access to multiple events matched by these new quantifiers? The `PatternSelectFunction.select` takes an argument of `Map<String,IN>` and not `Map<String,IN[]>`.
Which is the one being passed to `select` in the `Map`? The last matched? The first matched?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105698031
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -71,19 +70,16 @@
private final NonDuplicatingTypeSerializer<T> nonDuplicatingTypeSerializer;
// Buffer used to store the matched events
--- End diff --
Why `String` and not `State`? This may have problems with backwards compatibility.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105857677
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
+ 5.0;
}
})
- .followedBy("middle-second").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ .followedBy("middle-second").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getName().equals("next-one");
+ return value.getName()
+ .equals(
+ "next-one");
}
})
.followedBy("end").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
public boolean filter(Event value) throws Exception {
- return value.getName().equals("end");
+ return value.getName()
+ .equals("end");
}
});
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
List<Map<String, Event>> resultingPatterns = new ArrayList<>();
- for (StreamRecord<Event> inputEvent: inputEvents) {
+ for (StreamRecord<Event> inputEvent : inputEvents) {
--- End diff --
Reverted
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106163471
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
--- End diff --
I've done it the other way round as to handle optional in one place. But I think your version is easier to analyze.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106107521
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
--- End diff --
Here I have 3 comments:
1) you can create local variables instead of reusing `sourceState` and `sinkState`.
2) you can add a check like `Preconditions.checkArgument(currentPattern.getQuantifier() == Quantifier.TIMES)`
3) currently we cannot have `OPTIONAL` and `TIMES` but the code seems to support that. Am I correct? In this case we could also have in the `Pattern` a method called `atMost(int times)` which will allow the user to specify such a pattern. What do you think?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854452
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105856821
--- Diff: flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java ---
@@ -272,37 +372,42 @@ public void testBranchingPattern() {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- })
- .followedBy("middle-first").subtype(SubEvent.class).where(new FilterFunction<SubEvent>() {
+ }).followedBy("middle-first").subtype(SubEvent.class)
+ .where(new FilterFunction<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
- return value.getVolume() > 5.0;
+ return value.getVolume() >
--- End diff --
I reverted the whole `testSimplePatternNFA`. Probably I've run formatter on whole file.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys closed the pull request at:
https://github.com/apache/flink/pull/3477
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by eliaslevy <gi...@git.apache.org>.
Github user eliaslevy commented on the issue:
https://github.com/apache/flink/pull/3477
I created [FLINK-6419](https://issues.apache.org/jira/browse/FLINK-6419).
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
Hi, I've addressed the style comments. Still have some doubts about the `Preconditions`. I will try to add more comments on the `NFACompiler` with the implementation that will handle the failing test case. So I believe it will be okay to wait with further reviewing until then.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105699988
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
--- End diff --
Move the `resultingComputationStates` just before the `for (StateTransition<T> edge : edges) {...}`.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105700356
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
--- End diff --
This can become: `return takeBranches == 0 && ignoreBranches == 0 ? 0 : ignoreBranches + 1;`
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105855001
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
+ return 0;
+ }
+
+ return ignoreBranches + 1;
+ }
+
+ private ComputationState<T> createStartState(final ComputationState<T> computationState,
+ final int totalBranches) {
+ final DeweyNumber startVersion = computationState.getVersion().increase(totalBranches);
- boolean branched = false;
+ return ComputationState.createStartState(
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854576
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106150129
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java ---
@@ -74,11 +80,35 @@ public long getStartTimestamp() {
return state;
}
+ public State<T> getPreviousState() {
+ return previousState;
+ }
+
public T getEvent() {
return event;
}
public DeweyNumber getVersion() {
return version;
}
+
+ public static <T> ComputationState<T> createStartState(final State<T> state) {
+ Preconditions.checkArgument(state.isStart());
+ return new ComputationState<>(state, null, null, -1L, new DeweyNumber(1), -1L);
+ }
+
+ public static <T> ComputationState<T> createStartState(final State<T> state, final DeweyNumber version) {
+ Preconditions.checkArgument(state.isStart());
+ return new ComputationState<>(state, null, null, -1L, version, -1L);
+ }
+
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Sounds good! Thanks @dawidwys
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
I've updated the PR with a solution for the previously failing test. Also tried to add some more comments. @kl0u I think it is ready for another round of review :)
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
Thanks @kl0u for your comments, I've changed all code style specific.
Some comments were addressing a behaviour that one `Pattern` can be translated into multiple "equivalent" `States` with the same name.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105700521
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -247,27 +275,148 @@ public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
* @return Collection of computation states which result from the current one
*/
private Collection<ComputationState<T>> computeNextStates(
- final ComputationState<T> computationState,
- final T event,
- final long timestamp) {
- Stack<State<T>> states = new Stack<>();
- ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
- State<T> state = computationState.getState();
+ final ComputationState<T> computationState,
+ final T event,
+ final long timestamp) {
+ final ArrayList<ComputationState<T>> resultingComputationStates = new ArrayList<>();
+
+ final OutgoingEdges<T> outgoingEdges = createDecisionGraph(computationState, event);
+
+ // Create the computing version based on the previously computed edges
+ // We need to defer the creation of computation states until we know how many edges start
+ // at this computation state so that we can assign proper version
+ final List<StateTransition<T>> edges = outgoingEdges.getEdges();
+ Integer takeBranchesToVisit = Math.max(0, outgoingEdges.getTotalTakeBranches() - 1);
+ Integer ignoreBranchesToVisit = outgoingEdges.getTotalIgnoreBranches();
+ for (StateTransition<T> edge : edges) {
+ switch (edge.getAction()) {
+ case IGNORE: {
+ if (!computationState.isStartState()) {
+ final DeweyNumber version;
+ if (!isEquivalentState(edge.getTargetState(), computationState.getState())) {
+ version = computationState.getVersion().increase(ignoreBranchesToVisit).addStage();
+ ignoreBranchesToVisit--;
+ } else {
+ final int toIncrease = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(),
+ outgoingEdges.getTotalTakeBranches());
+ version = computationState.getVersion().increase(toIncrease);
+ }
+
+ resultingComputationStates.add(
+ ComputationState.createState(
+ edge.getTargetState(),
+ computationState.getPreviousState(),
+ computationState.getEvent(),
+ computationState.getTimestamp(),
+ version,
+ computationState.getStartTimestamp()
+ )
+ );
+ sharedBuffer.lock(
+ edge.getTargetState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+ }
+ break;
+ case TAKE:
+ final State<T> newState = edge.getTargetState();
+ final State<T> consumingState = edge.getSourceState();
+ final State<T> previousEventState = computationState.getPreviousState();
+
+ final T previousEvent = computationState.getEvent();
+ final DeweyNumber currentVersion = computationState.getVersion();
+
+ final DeweyNumber newComputationStateVersion = new DeweyNumber(currentVersion).addStage().increase(takeBranchesToVisit);
+ takeBranchesToVisit--;
+
+ final long startTimestamp;
+ if (computationState.isStartState()) {
+ startTimestamp = timestamp;
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ currentVersion);
+ } else {
+ startTimestamp = computationState.getStartTimestamp();
+ sharedBuffer.put(
+ consumingState.getName(),
+ event,
+ timestamp,
+ previousEventState.getName(),
+ previousEvent,
+ computationState.getTimestamp(),
+ currentVersion);
+ }
+
+ // a new computation state is referring to the shared entry
+ sharedBuffer.lock(consumingState.getName(), event, timestamp);
+
+ resultingComputationStates.add(ComputationState.createState(
+ newState,
+ consumingState,
+ event,
+ timestamp,
+ newComputationStateVersion,
+ startTimestamp
+ ));
+ break;
+ }
+ }
- states.push(state);
+ if (computationState.isStartState()) {
+ final int totalBranches = calculateIncreasingSelfState(outgoingEdges.getTotalIgnoreBranches(), outgoingEdges.getTotalTakeBranches());
+ final ComputationState<T> startState = createStartState(computationState, totalBranches);
+ resultingComputationStates.add(startState);
+ }
+
+ if (computationState.getEvent() != null) {
+ // release the shared entry referenced by the current computation state.
+ sharedBuffer.release(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ // try to remove unnecessary shared buffer entries
+ sharedBuffer.remove(
+ computationState.getState().getName(),
+ computationState.getEvent(),
+ computationState.getTimestamp());
+ }
+
+ return resultingComputationStates;
+ }
+
+ private int calculateIncreasingSelfState(int ignoreBranches, int takeBranches) {
+ if (takeBranches == 0 && ignoreBranches == 0) {
+ return 0;
+ }
+
+ return ignoreBranches + 1;
+ }
+
+ private ComputationState<T> createStartState(final ComputationState<T> computationState,
+ final int totalBranches) {
+ final DeweyNumber startVersion = computationState.getVersion().increase(totalBranches);
- boolean branched = false;
+ return ComputationState.createStartState(
--- End diff --
Move arguments in the same line
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Hi @dawidwys , I have 2 comments for this PR:
1) in the `findFinalStateAfterProceed()` in the `NFA` you should also check the filter condition as you traverse the `PROCEED` edges towards the end.
2) In this PR the computation states are consuming, and the `endState` is a "dummy" state.
Given this, in the `extractPatternMatches()` in the NFA, you do a
```
sharedBuffer.extractPatterns(
computationState.getPreviousState().getName(),
computationState.getEvent(),
computationState.getTimestamp(),
computationState.getVersion());
```
to extract the matched patterns.
This will work for patterns that terminate properly, as the `endState` now is a "dummy" state and you do not want to present it. But if I understand correctly, for timed out patterns, this means that the last state will not be presented. Is this correct?
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106150181
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106142265
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
--- End diff --
I agree that it is defensive. The only reason is that:
1) why not being defensive on these things? :P
2) I believe the code is more readable if the assumptions are clear when you start reading a method (although also the name can play the same role). If you see this check, then you know what is happening.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105855825
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/StateTransition.java ---
@@ -55,6 +65,7 @@ public boolean equals(Object obj) {
StateTransition<T> other = (StateTransition<T>) obj;
--- End diff --
Done
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on the issue:
https://github.com/apache/flink/pull/3477
Right now if the pattern name is "a" the events will be returned with keys: "a[0]", "a[1]" and so on, but agree it may be counterintuitive. Please file a JIRA for 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 issue #3477: [Flink-3318][cep] Add support for quantifiers to CEP's pa...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3477
Thanks @dawidwys I will have a look now.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105854116
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -209,33 +204,66 @@ public boolean equals(Object obj) {
return nonDuplicatingTypeSerializer.equals(other.nonDuplicatingTypeSerializer) &&
sharedBuffer.equals(other.sharedBuffer) &&
states.equals(other.states) &&
- windowTime == other.windowTime &&
- startEventCounter == other.startEventCounter;
+ windowTime == other.windowTime;
} else {
return false;
}
}
@Override
public int hashCode() {
- return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime, startEventCounter);
+ return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime);
+ }
+
--- End diff --
It is needed to tell if `IGNORE` is self ignore meaning we stay e.g. in a looping state. Looping state can consist of two states with one not containing `PROCEED` edges.
So this method is required just in the `IGNORE` context, so I think it does not fit into the `State` class.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by dawidwys <gi...@git.apache.org>.
Github user dawidwys commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r106141824
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java ---
@@ -74,88 +76,233 @@
*/
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
- Pattern<T, ?> pattern,
- TypeSerializer<T> inputTypeSerializer,
+ final Pattern<T, ?> pattern,
+ final TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
// return a factory for empty NFAs
- return new NFAFactoryImpl<T>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
+ return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
} else {
- // set of all generated states
- Map<String, State<T>> states = new HashMap<>();
- long windowTime;
+ final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
+ nfaFactoryCompiler.compileFactory();
+ return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
+ }
+ }
- // this is used to enforse pattern name uniqueness.
- Set<String> patternNames = new HashSet<>();
+ private static class NFAFactoryCompiler<T> {
- Pattern<T, ?> succeedingPattern;
- State<T> succeedingState;
- Pattern<T, ?> currentPattern = pattern;
+ private final Set<String> usedNames = new HashSet<>();
+ private final List<State<T>> states = new ArrayList<>();
+ private long windowTime = 0;
+ private Pattern<T, ?> currentPattern;
+
+ NFAFactoryCompiler(final Pattern<T, ?> pattern) {
+ this.currentPattern = pattern;
+ }
+
+ /**
+ * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
+ * multiple NFAs.
+ */
+ void compileFactory() {
// we're traversing the pattern from the end to the beginning --> the first state is the final state
- State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
- patternNames.add(currentPattern.getName());
+ State<T> sinkState = createEndingState();
+ // add all the normal states
+ sinkState = createMiddleStates(sinkState);
+ // add the beginning state
+ createStartState(sinkState);
+ }
+
+ List<State<T>> getStates() {
+ return states;
+ }
- states.put(currentPattern.getName(), currentState);
+ long getWindowTime() {
+ return windowTime;
+ }
+
+ private State<T> createEndingState() {
+ State<T> sinkState = new State<>(ENDING_STATE_NAME, State.StateType.Final);
+ states.add(sinkState);
+ usedNames.add(ENDING_STATE_NAME);
windowTime = currentPattern.getWindowTime() != null ? currentPattern.getWindowTime().toMilliseconds() : 0L;
+ return sinkState;
+ }
- while (currentPattern.getPrevious() != null) {
- succeedingPattern = currentPattern;
- succeedingState = currentState;
- currentPattern = currentPattern.getPrevious();
+ private State<T> createMiddleStates(final State<T> sinkState) {
- if (!patternNames.add(currentPattern.getName())) {
- throw new MalformedPatternException("Duplicate pattern name: " + currentPattern.getName() + ". " +
- "Pattern names must be unique.");
+ State<T> lastSink = sinkState;
+ while (currentPattern.getPrevious() != null) {
+ State<T> sourceState;
+
+ checkPatternNameUniqueness();
+
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ convertToLooping(lastSink, sourceState);
+
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ sourceState = createFirstMandatoryStateOfLoop(
+ sourceState,
+ State.StateType.Normal
+ );
+ states.add(sourceState);
+ usedNames.add(sourceState.getName());
+ }
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ sourceState = convertToTimesState(lastSink, sourceState, currentPattern.getTimes());
+ } else {
+ convertToSingletonState(
+ lastSink,
+ sourceState);
}
- Time currentWindowTime = currentPattern.getWindowTime();
+ currentPattern = currentPattern.getPrevious();
+ lastSink = sourceState;
+ final Time currentWindowTime = currentPattern.getWindowTime();
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
// the window time is the global minimum of all window times of each state
windowTime = currentWindowTime.toMilliseconds();
}
+ }
+
+ return lastSink;
+ }
- if (states.containsKey(currentPattern.getName())) {
- currentState = states.get(currentPattern.getName());
+ private void checkPatternNameUniqueness() {
+ if (usedNames.contains(currentPattern.getName())) {
+ throw new MalformedPatternException(
+ "Duplicate pattern name: " + currentPattern.getName() + ". " +
+ "Pattern names must be unique.");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createStartState(State<T> sinkState) {
+ final State<T> beginningState;
+
+ checkPatternNameUniqueness();
+
+ if (currentPattern.getQuantifier().isLooping()) {
+ final State<T> loopingState;
+ if (currentPattern.getQuantifier().isAtLeastOne()) {
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ beginningState = createFirstMandatoryStateOfLoop(loopingState, State.StateType.Start);
+ states.add(loopingState);
} else {
- currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
- states.put(currentState.getName(), currentState);
+ loopingState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState = loopingState;
}
-
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.TAKE,
- succeedingState,
- (FilterFunction<T>) succeedingPattern.getFilterFunction()));
-
- if (succeedingPattern instanceof FollowedByPattern) {
- // the followed by pattern entails a reflexive ignore transition
- currentState.addStateTransition(new StateTransition<T>(
- StateTransitionAction.IGNORE,
- currentState,
- null
- ));
+ convertToLooping(sinkState, loopingState);
+ } else if (currentPattern.getQuantifier() == Quantifier.TIMES) {
+ if (currentPattern.getTimes() > 1) {
+ final State<T> timesState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(timesState);
+ sinkState = convertToTimesState(sinkState, timesState, currentPattern.getTimes() - 1);
}
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
+ } else {
+ beginningState = new State<>(currentPattern.getName(), State.StateType.Start);
+ beginningState.addTake(sinkState, (FilterFunction<T>) currentPattern.getFilterFunction());
}
- // add the beginning state
- final State<T> beginningState;
+ states.add(beginningState);
+ usedNames.add(beginningState.getName());
+
+ return beginningState;
+ }
+
+ private State<T> convertToTimesState(State<T> sinkState, State<T> sourceState, int times) {
+ convertToSingletonState(sinkState, sourceState);
+ for (int i = 0; i < times - 1; i++) {
+ sinkState = sourceState;
+ sourceState = new State<>(currentPattern.getName(), State.StateType.Normal);
+ states.add(sourceState);
+ convertToSingletonState(sinkState, sourceState);
+ }
+ return sourceState;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void convertToSingletonState(
+ final State<T> sinkState,
+ final State<T> sourceState) {
+
+ final FilterFunction<T> currentFilterFunction = (FilterFunction<T>) currentPattern.getFilterFunction();
+ final FilterFunction<T> trueFunction = FilterFunctions.trueFunction();
+ sourceState.addTake(sinkState, currentFilterFunction);
+
+ final State<T> ignoreState;
+ if (currentPattern.getQuantifier() == Quantifier.OPTIONAL) {
+ sourceState.addProceed(sinkState, trueFunction);
+ ignoreState = new State<>(currentPattern.getName(), State.StateType.Normal);
- if (states.containsKey(BEGINNING_STATE_NAME)) {
- beginningState = states.get(BEGINNING_STATE_NAME);
+ ignoreState.addTake(sinkState, currentFilterFunction);
+ states.add(ignoreState);
} else {
- beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
- states.put(BEGINNING_STATE_NAME, beginningState);
+ ignoreState = sourceState;
+ }
+
+ if (currentPattern instanceof FollowedByPattern) {
+ sourceState.addIgnore(ignoreState, trueFunction);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private State<T> createFirstMandatoryStateOfLoop(
+ final State<T> sinkState,
+ final State.StateType stateType) {
+
--- End diff --
I think it is an internal method in internal class. Not sure if we should be so defensive. If the check fails it means some error in implemented algorithm not in provided arguments.
---
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 #3477: [Flink-3318][cep] Add support for quantifiers to C...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3477#discussion_r105699441
--- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -209,33 +204,66 @@ public boolean equals(Object obj) {
return nonDuplicatingTypeSerializer.equals(other.nonDuplicatingTypeSerializer) &&
sharedBuffer.equals(other.sharedBuffer) &&
states.equals(other.states) &&
- windowTime == other.windowTime &&
- startEventCounter == other.startEventCounter;
+ windowTime == other.windowTime;
} else {
return false;
}
}
@Override
public int hashCode() {
- return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime, startEventCounter);
+ return Objects.hash(nonDuplicatingTypeSerializer, sharedBuffer, states, windowTime);
+ }
+
+ private static <T> boolean isEquivalentState(final State<T> s1, final State<T> s2) {
+ return s1.getName().equals(s2.getName());
}
/**
- * Comparator used for imposing the assumption that IGNORE is always the last StateTransition in a state.
- */
- private interface StateTransitionComparator<T> extends Comparator<StateTransition<T>>, Serializable {}
- private final Comparator<StateTransition<T>> stateTransitionComparator = new StateTransitionComparator<T>() {
- private static final long serialVersionUID = -2775474935413622278L;
+ * Structures to keep decisions based on the transition actions, that counts the number of taken actions.
+ */
+ private static class OutgoingEdges<T> {
+ private List<StateTransition<T>> edges = new ArrayList<>();
- @Override
- public int compare(final StateTransition<T> o1, final StateTransition<T> o2) {
- if (o1.getAction() == o2.getAction()) {
- return 0;
+ private final State<T> currentState;
+
+ private int totalTakeBranches = 0;
+ private int totalIgnoreBranches = 0;
+
+ OutgoingEdges(final State<T> currentState) {
+ this.currentState = currentState;
+ }
+
+ void add(StateTransition<T> edge) {
+
+ if (!isSelfIgnore(edge)) {
+ if (edge.getAction() == StateTransitionAction.IGNORE) {
+ totalIgnoreBranches++;
+ } else if (edge.getAction() == StateTransitionAction.TAKE) {
+ totalTakeBranches++;
+ }
}
- return o1.getAction() == StateTransitionAction.IGNORE ? 1 : -1;
+
+ edges.add(edge);
+ }
+
+ int getTotalIgnoreBranches() {
+ return totalIgnoreBranches;
+ }
+ int getTotalTakeBranches() {
+ return totalTakeBranches;
}
- };
+
+ List<StateTransition<T>> getEdges() {
+ return edges;
+ }
+
--- End diff --
Why is this method needed? The two end-points of the IGNORE edge, won't they be the same state?
---
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.
---