[FLINK-8914] CEP's greedy() modifier doesn't work - ASF JIRA

Agile Board

Rank to Top

Rank to Bottom

Attach files

Attach Screenshot

Bulk Copy Attachments

Bulk Move Attachments

Voters

Stop watching

Watchers

Create sub-task

Convert to sub-task

Link

Clone

Labels

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Closed
Priority: Major
Resolution: Not A Problem
Affects Version/s: 1.4.0, 1.4.1
Fix Version/s: None
Component/s: Library / CEP
Labels:
- pull-request-available

Description

When applied to the first or last component of a CEP Pattern, greedy() doesn't work correctly. Here's an example:

package com.dataartisans.flinktraining.exercises.datastream_java.cep;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

public class RunLength {

  public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    DataStream<Integer> input = env.fromElements(1, 1, 1, 1, 1, 0, 1, 1, 1, 0);

    Pattern<Integer, ?> onesThenZero = Pattern.<Integer>begin("ones")
      .where(new SimpleCondition<Integer>() {
        @Override
        public boolean filter(Integer value) throws Exception {
          return value == 1;
        }
      })
      .oneOrMore()
      .greedy()
      .consecutive()
      .next("zero")
      .where(new SimpleCondition<Integer>() {
        @Override
        public boolean filter(Integer value) throws Exception {
          return value == 0;
        }
      });

      PatternStream<Integer> patternStream = CEP.pattern(input, onesThenZero);

      // Expected: 5 3
      // Actual: 5 4 3 2 1 3 2 1
      patternStream.select(new LengthOfRun()).print();

      env.execute();
    }

    public static class LengthOfRun implements PatternSelectFunction<Integer, Integer> {
      public Integer select(Map<String, List<Integer>> pattern) {
      return pattern.get("ones").size();
    }
  }
}

The only workaround for now seems to be to rewrite the pattern so that greedy() isn't needed – i.e. by bracketing the greedy section with a prefix and suffix that both have to be matched.