dianfu commented on code in PR #20029: URL: https://github.com/apache/flink/pull/20029#discussion_r906937708
########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java: ########## @@ -579,6 +662,21 @@ private void checkIfPreviousPatternGreedy() { } } + private void checkWindowTimeBetweenEvents(Time windowTime, WithinType withinType) { Review Comment: What about moving this validation to NFAFactoryCompilter.compileFactory? It window time of `FIRST_AND_LAST` may be defined in other Pattern object which is only available during compiling the Pattern. ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java: ########## @@ -394,14 +447,28 @@ public Pattern<T, F> times(int times) { * @throws MalformedPatternException if the quantifier is not applicable to this pattern. */ public Pattern<T, F> times(int from, int to) { + return times(from, to, null); + } + + /** + * Specifies that the pattern can occur between from and to times with time interval corresponds + * to the maximum time gap between previous and current event for each times. + * + * @param from number of times matching event must appear at least + * @param to number of times matching event must appear at most + * @param windowTime time of the matching window between times + * @return The same pattern with the number of times range applied + * @throws MalformedPatternException if the quantifier is not applicable to this pattern. + */ + public Pattern<T, F> times(int from, int to, Time windowTime) { Review Comment: ```suggestion public Pattern<T, F> times(int from, int to, @Nullable Time windowTime) { ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java: ########## @@ -206,12 +209,16 @@ public int getTo() { return to; } - public static Times of(int from, int to) { - return new Times(from, to); + public Time getWindowTime() { + return windowTime; } - public static Times of(int times) { - return new Times(times, times); + public static Times of(int from, int to, Time windowTime) { Review Comment: ```suggestion public static Times of(int from, int to, @Nullable Time windowTime) { ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java: ########## @@ -137,6 +144,8 @@ public static boolean canProduceEmptyMatches(final Pattern<?, ?> pattern) { private final NFAStateNameHandler stateNameHandler = new NFAStateNameHandler(); private final Map<String, State<T>> stopStates = new HashMap<>(); private final List<State<T>> states = new ArrayList<>(); + private final Map<String, Integer> ignoreTimes = new HashMap<>(); Review Comment: ignoreTimes is not used ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/MigrationUtils.java: ########## @@ -112,6 +112,7 @@ static <T> Queue<ComputationState> deserializeComputationStates( long timestamp = timestampSerializer.deserialize(source); DeweyNumber version = versionSerializer.deserialize(source); long startTimestamp = timestampSerializer.deserialize(source); + long previousTimestamp = timestampSerializer.deserialize(source); Review Comment: ```suggestion long previousTimestamp = -1; ``` This is to allow migrating state from checkpoint/savepoint generated prior to Flink 1.5 and so I guess we could not just deserialize from it. ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java: ########## @@ -41,6 +41,9 @@ public class ComputationState { // Timestamp of the first element in the pattern private final long startTimestamp; + // Timestamp of the previous element in the state Review Comment: ```suggestion // Timestamp of the previous element of the pattern ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java: ########## @@ -414,10 +481,26 @@ public Pattern<T, F> times(int from, int to) { * @throws MalformedPatternException if the quantifier is not applicable to this pattern. */ public Pattern<T, F> timesOrMore(int times) { + return timesOrMore(times, null); + } + + /** + * Specifies that this pattern can occur the specified times at least with interval corresponds + * to the maximum time gap between previous and current event for each times. This means at + * least the specified times and at most infinite number of events can be matched to this + * pattern. + * + * @param times number of times at least matching event must appear + * @param windowTime time of the matching window between times + * @return The same pattern with a {@link Quantifier#looping(ConsumingStrategy)} quantifier + * applied. + * @throws MalformedPatternException if the quantifier is not applicable to this pattern. + */ + public Pattern<T, F> timesOrMore(int times, Time windowTime) { Review Comment: ```suggestion public Pattern<T, F> timesOrMore(int times, @Nullable Time windowTime) { ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java: ########## @@ -223,12 +230,12 @@ public boolean equals(Object o) { return false; } Times times = (Times) o; - return from == times.from && to == times.to; + return from == times.from && to == times.to && windowTime == times.windowTime; Review Comment: ```suggestion return from == times.from && to == times.to && windowTime.toMilliseconds() == times.windowTime.toMilliseconds(); ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java: ########## @@ -377,11 +417,24 @@ public Pattern<T, F> greedy() { * @throws MalformedPatternException if the quantifier is not applicable to this pattern. */ public Pattern<T, F> times(int times) { + return times(times, null); + } + + /** + * Specifies exact number of times that this pattern should be matched and time interval + * corresponds to the maximum time gap between previous and current event for each times. + * + * @param times number of times matching event must appear + * @param windowTime time of the matching window between times + * @return The same pattern with number of times applied + * @throws MalformedPatternException if the quantifier is not applicable to this pattern. + */ + public Pattern<T, F> times(int times, Time windowTime) { Review Comment: ```suggestion public Pattern<T, F> times(int times, @Nullable Time windowTime) { ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializerSnapshot.java: ########## @@ -67,6 +72,15 @@ protected NFAStateSerializer createOuterSerializerWithNestedSerializers( @SuppressWarnings("unchecked") TypeSerializer<EventId> eventIdSerializer = (TypeSerializer<EventId>) nestedSerializers[2]; - return new NFAStateSerializer(versionSerializer, nodeIdSerializer, eventIdSerializer); + return new NFAStateSerializer( + versionSerializer, nodeIdSerializer, eventIdSerializer, supportsPreviousTimestamp); + } + + @Override + protected void readOuterSnapshot( + int readOuterSnapshotVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException { + super.readOuterSnapshot(readOuterSnapshotVersion, in, userCodeClassLoader); + supportsPreviousTimestamp = readOuterSnapshotVersion == CURRENT_VERSION; Review Comment: Add static variable `FIRST_VERSION_WITH_PREVIOUS_TIMESTAMP` and change this to the following: ``` @Override protected void readOuterSnapshot( int readOuterSnapshotVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException { if (readOuterSnapshotVersion < FIRST_VERSION_WITH_PREVIOUS_TIMESTAMP) { supportsPreviousTimestamp = false; } else if (readOuterSnapshotVersion == FIRST_VERSION_WITH_PREVIOUS_TIMESTAMP) { supportsPreviousTimestamp = true; } else { supportsPreviousTimestamp = in.readBoolean(); } } ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java: ########## @@ -187,15 +188,17 @@ public enum ConsumingStrategy { public static class Times { private final int from; private final int to; + private final Time windowTime; Review Comment: ```suggestion private final @Nullable Time windowTime; ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java: ########## @@ -273,7 +282,21 @@ public Collection<Map<String, List<T>>> process( new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR); for (ComputationState computationState : nfaState.getPartialMatches()) { - if (isStateTimedOut(computationState, timestamp)) { + String currentStateName = computationState.getCurrentStateName(); + boolean withinStateTimeout = Review Comment: The name *withinStateTimeout* and *isStateTimeout* are not quite clear. What about finding some other name, e.g. isStateTimeoutBetweenEvents vs isStateTimeoutWholePattern ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java: ########## @@ -348,10 +370,28 @@ public Pattern<T, F> optional() { * @throws MalformedPatternException if the quantifier is not applicable to this pattern. */ public Pattern<T, F> oneOrMore() { + return oneOrMore(null); + } + + /** + * Specifies that this pattern can occur {@code one or more} times and time interval corresponds + * to the maximum time gap between previous and current event for each times. This means at + * least one and at most infinite number of events can be matched to this pattern. + * + * <p>If this quantifier is enabled for a pattern {@code A.oneOrMore().followedBy(B)} and a + * sequence of events {@code A1 A2 B} appears, this will generate patterns: {@code A1 B} and + * {@code A1 A2 B}. See also {@link #allowCombinations()}. + * + * @param windowTime time of the matching window between times + * @return The same pattern with a {@link Quantifier#looping(ConsumingStrategy)} quantifier + * applied. + * @throws MalformedPatternException if the quantifier is not applicable to this pattern. + */ + public Pattern<T, F> oneOrMore(Time windowTime) { Review Comment: ```suggestion public Pattern<T, F> oneOrMore(@Nullable Time windowTime) { ``` ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java: ########## @@ -390,10 +405,28 @@ private State<T> convertPattern(final State<T> sinkState) { return lastSink; } + private State<T> createState(State.StateType stateType, boolean notIgnore) { + State<T> state = createState(currentPattern.getName(), stateType); + if (notIgnore) { Review Comment: What does the variable name *notIgnore* mean? It's not quite clear for me. ########## flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializerSnapshot.java: ########## @@ -67,6 +72,15 @@ protected NFAStateSerializer createOuterSerializerWithNestedSerializers( @SuppressWarnings("unchecked") TypeSerializer<EventId> eventIdSerializer = (TypeSerializer<EventId>) nestedSerializers[2]; - return new NFAStateSerializer(versionSerializer, nodeIdSerializer, eventIdSerializer); + return new NFAStateSerializer( + versionSerializer, nodeIdSerializer, eventIdSerializer, supportsPreviousTimestamp); + } Review Comment: I guess we need implement the following method to allow restoring state generated from previous version: ``` @Override protected OuterSchemaCompatibility resolveOuterSchemaCompatibility( NFAStateSerializer newSerializer) { if (supportsPreviousTimestamp != newSerializer.supportsPreviousTimestamp) { return OuterSchemaCompatibility.COMPATIBLE_AFTER_MIGRATION; } return OuterSchemaCompatibility.COMPATIBLE_AS_IS; } ``` Besides, we should also implement the following method: ``` @Override protected void writeOuterSnapshot(DataOutputView out) throws IOException { out.writeBoolean(supportsPreviousTimestamp); } ``` ########## flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesOrMoreITCase.java: ########## @@ -37,6 +41,14 @@ /** Tests for {@link Pattern#timesOrMore(int)}. */ public class TimesOrMoreITCase extends TestLogger { + + @Parameterized.Parameter public Time time; + + @Parameterized.Parameters(name = "Times Range Time: {0}") Review Comment: The results for the following two cases are same? Is this expected? ########## flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java: ########## @@ -1560,16 +1570,16 @@ public boolean filter(Event value) throws Exception { } @Test - public void testNotFollowByBeforeTimesWithIn() throws Exception { + public void testNotFollowByBeforeTimesWithin() throws Exception { List<StreamRecord<Event>> inputEvents = new ArrayList<>(); Event a1 = new Event(40, "a", 1.0); Event b1 = new Event(41, "b", 2.0); Event a2 = new Event(42, "a", 3.0); Event c1 = new Event(43, "c", 4.0); Event c2 = new Event(44, "c", 5.0); - Event a3 = new Event(46, "a", 7.0); Review Comment: Why changed this test case? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org