sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/nfa/NFA.java

/*
 * 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.beam.sdk.extensions.sql.impl.nfa;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPCall;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPFieldRef;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPKind;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPLiteral;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPOperation;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPOperator;
import org.apache.beam.sdk.extensions.sql.impl.cep.CEPPattern;
import org.apache.beam.sdk.extensions.sql.impl.cep.Quantifier;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.values.Row;

// TODO: add support for more quantifiers: `?`, `{19, }` ... for now, support `+` and singletons
// TODO: sort conditions based on "the last identifier" during compilation
// TODO: add optimization for the NFA
// for now, assume the conditions are sorted.

/**
 * {@code NFA} is an implementation of non-deterministic finite automata. It is used for the
 * non-deterministic pattern match in the MATCH_RECOGNIZE component.
 *
 * <p>The implementation is strongly based on the UMASS paper on NFA with shared buffer:
 *
 * @see <a
 *     href="https://dl.acm.org/doi/10.1145/1376616.1376634">https://dl.acm.org/doi/10.1145/1376616.1376634</a>
 */
@SuppressWarnings({
  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
})
public class NFA implements Serializable {

  private final State startState;
  private ArrayList<StateLocator> currentRuns;
  private final Schema upstreamSchema;

  private NFA(List<CEPPattern> patterns, Schema upstreamSchema) {
    this.startState = loadStates(patterns);
    this.currentRuns = new ArrayList<>();
    this.upstreamSchema = upstreamSchema;
  }

  public static NFA compile(List<CEPPattern> patterns, Schema outSchema) {
    return new NFA(patterns, outSchema);
  }

  // process a new row,
  // return a mapping of pattern variable to row to output in the DoFn if there is a match
  // return null if none of locators reached the final state
  public Map<String, ArrayList<Row>> processNewRow(Row inputRow) {
    // wrap the input row as an event
    Event inputEvent = new Event(inputRow, upstreamSchema);
    ArrayList<StateLocator> nextStateLocators = new ArrayList<>();
    // add a start state locator to the array
    EventPointer nullPtr = new EventPointer(new ArrayList<>(), "");
    currentRuns.add(new StateLocator(nullPtr, startState, startState, 0, null));
    // scan for kleene plus locator, if exits, add the next next state
    ArrayList<StateLocator> kleenePlusLocators = new ArrayList<>();
    for (StateLocator locator : currentRuns) {
      if (locator.isKleenePlusSecondary()) {
        if (locator.getCurState().getNextState().isFinal) {
          return processOutput(locator);
        }
        StateLocator proceedLocator = locator.proceedIgnore();
        kleenePlusLocators.add(proceedLocator);
      }
    }
    currentRuns.addAll(kleenePlusLocators);

    // decide next possible states
    for (StateLocator currentLocator : currentRuns) {
      boolean split = false;
      StateLocator proceedLocator = currentLocator.proceed(inputEvent);
      if (proceedLocator != null) {
        // TODO: add support for after match strategy
        // if the locator is at the final state
        // clear current runs
        // then return the output
        if (proceedLocator.atFinal()) {
          // reset after a match
          currentRuns.clear();
          State iterState = startState;
          while (!iterState.isFinal) {
            iterState.reset();
            iterState = iterState.getNextState();
          }
          currentRuns.add(new StateLocator(nullPtr, startState, startState, 0, null));

          return processOutput(proceedLocator);
        } else {
          // add the new locator to the array
          nextStateLocators.add(proceedLocator);
        }
        split = true;
      }
      StateLocator takeLocator = currentLocator.take(inputEvent, split);
      if (takeLocator != null) {
        nextStateLocators.add(takeLocator);
      }
    }
    this.currentRuns = nextStateLocators;
    return null;
  }

  // returns a row with the output schema once a locator reaches the final state
  private Map<String, ArrayList<Row>> processOutput(StateLocator locator) {
    HashMap<String, ArrayList<Row>> rows = new HashMap<>();
    EventPointer iterPointer = locator.getPointer().copy();
    Event curEvent = locator.getCurrentEvent();

    while (curEvent != null) {
      String patternVar = iterPointer.getPatternVar();
      if (rows.containsKey(patternVar)) {
        ArrayList<Row> patternArray = rows.get(patternVar);
        patternArray.add(curEvent.getRow());
      } else {
        ArrayList<Row> newPatternArray = new ArrayList<>();
        newPatternArray.add(curEvent.getRow());
        rows.put(patternVar, newPatternArray);
      }
      curEvent = curEvent.findEvent(iterPointer);
      if (curEvent != null) {
        iterPointer = curEvent.getPrevPointer(iterPointer);
      }
    }

    // restore the order
    for (ArrayList<Row> i : rows.values()) {
      Collections.reverse(i);
    }

    return rows;
  }

  private static class Event implements Serializable {
    private HashMap<EventPointer, Event> prevEvents =
        new HashMap<>(); // a mapping from the runIndex to the previous events
    private Row row;
    private Schema upstreamSchema;

    Event(Row inputRow, Schema upstreamSchema) {
      this.row = inputRow;
      this.upstreamSchema = upstreamSchema;
    }

    // store an input row as a literal
    public CEPLiteral toCEPLiteral(CEPFieldRef fieldRef) {
      int fieldIndex = fieldRef.getIndex();
      Schema.Field field = upstreamSchema.getField(fieldIndex);
      Schema.FieldType type = field.getType();
      switch (type.getTypeName()) {
        case BYTE:
          return CEPLiteral.of(row.getByte(fieldIndex));
        case INT16:
          return CEPLiteral.of(row.getInt16(fieldIndex));
        case INT32:
          return CEPLiteral.of(row.getInt32(fieldIndex));
        case INT64:
          return CEPLiteral.of(row.getInt64(fieldIndex));
        case DECIMAL:
          return CEPLiteral.of(row.getDecimal(fieldIndex));
        case FLOAT:
          return CEPLiteral.of(row.getFloat(fieldIndex));
        case DOUBLE:
          return CEPLiteral.of(row.getDouble(fieldIndex));
        case DATETIME:
          return CEPLiteral.of(row.getDateTime(fieldIndex));
        case BOOLEAN:
          return CEPLiteral.of(row.getBoolean(fieldIndex));
        case STRING:
          return CEPLiteral.of(row.getString(fieldIndex));
        default:
          throw new UnsupportedOperationException(
              "The type is not supported: " + type.getTypeName().toString());
      }
    }

    private Event findEvent(EventPointer eventPointer) {
      return prevEvents.getOrDefault(eventPointer, null);
    }

    private EventPointer findEventPointer(EventPointer pointer) {
      for (EventPointer i : prevEvents.keySet()) {
        if (i.equals(pointer)) {
          return i.copy();
        }
      }
      return null;
    }

    public EventPointer getPrevPointer(EventPointer curPointer) {
      if (curPointer.isNull()) {
        return null;
      }
      if (curPointer.isProceedPointer()) {
        while (findEvent(curPointer) == null && curPointer.canTrim()) {
          curPointer.trim();
        }
        return findEventPointer(curPointer);
      } else {
        while (findEvent(curPointer) == null) {
          if (curPointer.canDecrement()) {
            curPointer.decrement();
          } else {
            return null;
          }
        }
        return curPointer;
      }
    }

    public Row getRow() {
      return row;
    }

    public void addPrevEvent(EventPointer ptr, @Nullable Event prevEvent) {
      prevEvents.put(ptr, prevEvent);
    }
  }

  // shared buffer versioned pointer: see the UMASS paper for description
  private static class EventPointer implements Serializable {
    private List<Integer> ptrValues;
    // labels the event that was pointed to
    private final String patternVar;

    EventPointer(List<Integer> ptrValues, String patternVar) {
      this.ptrValues = ptrValues;
      this.patternVar = patternVar;
    }

    public boolean canTrim() {
      return ptrValues.size() > 1;
    }

    // for following a proceed edge
    // trim the last digit
    public void trim() {
      if (isProceedPointer()) {
        ptrValues = ptrValues.subList(0, ptrValues.size() - 1);
      } else {
        throw new IllegalStateException("the null event pointer cannot be trimmed.");
      }
    }

    public boolean canDecrement() {
      if (isNull()) {
        return false;
      }
      int lastValue = ptrValues.get(ptrValues.size() - 1);
      return lastValue > 0;
    }

    // for following the take edge
    // decrement the last digit
    public void decrement() {
      if (!canDecrement()) {
        throw new IllegalStateException("the event pointer cannot be decremented.");
      }
      if (!isProceedPointer()) {
        ArrayList<Integer> newPtrValue =
            new ArrayList<>(ptrValues.subList(0, ptrValues.size() - 1));
        int lastValue = ptrValues.get(ptrValues.size() - 1) - 1;
        newPtrValue.add(lastValue);
        ptrValues = newPtrValue;
      } else {
        throw new IllegalStateException("the event pointer cannot be decremented.");
      }
    }

    @Override
    public boolean equals(Object other) {
      if (!(other instanceof EventPointer)) {
        return false;
      }
      EventPointer otherPointer = (EventPointer) other;
      return ptrValues.equals(otherPointer.ptrValues);
    }

    public boolean isProceedPointer() {
      // if the last digit of the pointer is 0,
      // then it is a proceed pointer
      if (ptrValues.isEmpty()) {
        return false;
      }
      int lastPtrValue = ptrValues.get(ptrValues.size() - 1);
      if (lastPtrValue == 0) {
        return true;
      } else {
        return false;
      }
    }

    @Override
    public int hashCode() {
      return ptrValues.hashCode();
    }

    public boolean isNull() {
      return ptrValues.isEmpty() && patternVar.equals("");
    }

    public String getPatternVar() {
      return patternVar;
    }

    public EventPointer copy() {
      ArrayList<Integer> newPtrValue = new ArrayList<>(ptrValues);
      return new EventPointer(newPtrValue, patternVar);
    }

    public EventPointer getNewProceedPointer(int value, String patternVar) {
      ArrayList<Integer> newPtrValue = new ArrayList<>(ptrValues);
      newPtrValue.add(value);
      return new EventPointer(newPtrValue, patternVar);
    }

    // for taking the take edge with splitting, set last pointer digit to the desired value
    public EventPointer getNewTakePointer(int value) {
      ArrayList<Integer> newPtrValue = new ArrayList<>(ptrValues);
      newPtrValue.set(newPtrValue.size() - 1, value);
      return new EventPointer(newPtrValue, patternVar);
    }

    @Override
    public String toString() {
      ArrayList<String> ptrValuesCopy = new ArrayList<>();
      for (Integer i : ptrValues) {
        ptrValuesCopy.add(i.toString());
      }
      return String.join(".", ptrValuesCopy);
    }
  }

  // StateLocator should be thought as immutable
  private static class StateLocator implements Serializable {
    private EventPointer ptr;
    private State startState;
    private State curState;
    private int takeCount = 0; // counts the number of events taken
    private Event curEvent = null;

    StateLocator(
        EventPointer ptr, State startState, State curState, int takeCount, Event curEvent) {
      this.ptr = ptr;
      this.startState = startState;
      this.curState = curState;
      this.takeCount = takeCount;
      this.curEvent = curEvent;
    }

    public State getCurState() {
      return curState;
    }

    public EventPointer getPointer() {
      return ptr;
    }

    public Event getCurrentEvent() {
      return curEvent;
    }

    // check if the current state is the final state
    public boolean atFinal() {
      return curState.isFinal;
    }

    public boolean isKleenePlusSecondary() {
      return curState.isKleenePlusSecondary();
    }

    // a proceed action for secondary kleenePlus state
    public StateLocator proceedIgnore() {
      if (isKleenePlusSecondary()) {
        EventPointer newPtr = ptr.getNewProceedPointer(0, curState.getPatternVar());
        return new StateLocator(newPtr, startState, curState.getNextState(), 0, curEvent);
      } else {
        return null;
      }
    }

    // represents the "proceed"/"begin" edge: transfer to a new state,
    // writes the new event in the new state's buffer.
    // returns a "new" state locator
    // returns null if not a match
    public StateLocator proceed(Event inputEvent) {
      if (curState.hasProceed()) {
        // try to proceed
        CEPCall condition = (CEPCall) curState.getProceedCondition();
        String patternVar = curState.getPatternVar();
        if (evalCondition(inputEvent, condition)) {
          // special case: for an event that starts a match,
          // assign a new index as the pointer value
          if (curState == startState && curEvent == null) {
            int ptrValue = curState.assignIndex();
            ArrayList<Integer> ptrArray = new ArrayList<>();
            ptrArray.add(ptrValue);
            EventPointer eventPointer = new EventPointer(ptrArray, patternVar);
            inputEvent.addPrevEvent(eventPointer, null);
            return new StateLocator(
                eventPointer, startState, curState.getNextState(), 0, inputEvent);
          }
          // for the other cases, add a zero in the event pointer
          EventPointer newPtr = ptr.getNewProceedPointer(0, patternVar);
          inputEvent.addPrevEvent(newPtr, curEvent);
          return new StateLocator(newPtr, startState, curState.getNextState(), 0, inputEvent);
        } else {
          return null;
        }
      } else {
        return null;
      }
    }

    // represents the "take" edge: the state stays at current state,
    // write the new event in the current state's buffer.
    // returns null if not a match
    public StateLocator take(Event inputEvent, boolean split) {
      if (curState.hasTake()) {
        if (evalCondition(inputEvent, curState.getTakeCondition())) {
          if (split) {
            // get another unique pointer value
            int ptrValue = curState.assignIndex();
            EventPointer newPtr = ptr.getNewTakePointer(ptrValue);
            inputEvent.addPrevEvent(newPtr, curEvent);
            return new StateLocator(newPtr, startState, curState, takeCount + 1, inputEvent);
          } else {
            EventPointer newPtr = ptr.copy();
            inputEvent.addPrevEvent(newPtr, curEvent);
            return new StateLocator(newPtr, startState, curState, takeCount + 1, inputEvent);
          }
        } else {
          return null;
        }
      } else {
        return null;
      }
    }

    // evaluates the condition
    // returns true if the new event satisfies the condition at a givens state
    private boolean evalCondition(Event inputEvent, CEPOperation operation) {
      if (operation == null) {
        return true;
      }
      CEPCall condition = (CEPCall) operation;
      CEPKind comparator = condition.getOperator().getCepKind();
      CEPOperation leftSide =
          condition.getOperands().get(0); // left side must contain field reference
      CEPOperation rightSide = condition.getOperands().get(1);
      CEPLiteral leftSideValue = evalLeftSideCondition(leftSide, inputEvent);
      CEPLiteral rightSideValue = evalRightSideCondition(rightSide, inputEvent);
      if (leftSideValue == null) {
        return false;
      }
      if (rightSideValue == null) {
        return true;
      }
      switch (comparator) {
        case EQUALS:
          return leftSideValue.compareTo(rightSideValue) == 0;
        case NOT_EQUALS:
          return leftSideValue.compareTo(rightSideValue) != 0;
        case GREATER_THAN:
          return leftSideValue.compareTo(rightSideValue) > 0;
        case GREATER_THAN_OR_EQUAL:
          return leftSideValue.compareTo(rightSideValue) >= 0;
        case LESS_THAN:
          return leftSideValue.compareTo(rightSideValue) < 0;
        case LESS_THAN_OR_EQUAL:
          return leftSideValue.compareTo(rightSideValue) <= 0;
        default:
          throw new IllegalStateException(
              "the comparator is not supported: " + comparator.toString());
      }
    }

    // evaluates the condition value (left) given an input event
    private CEPLiteral evalLeftSideCondition(CEPOperation inputOperation, Event inputEvent) {
      if (inputOperation instanceof CEPLiteral) {
        return (CEPLiteral) inputOperation;
      } else if (inputOperation.getClass() == CEPFieldRef.class) {
        return inputEvent.toCEPLiteral((CEPFieldRef) inputOperation);
      } else if (inputOperation.getClass() == CEPCall.class) {
        CEPCall call = (CEPCall) inputOperation;
        CEPOperator operator = call.getOperator();
        List<CEPOperation> operands = call.getOperands();
        switch (operator.getCepKind()) {
          case LAST:
            return last(
                operands.get(0), evalLeftSideCondition(operands.get(1), inputEvent), inputEvent);
          case PLUS:
            return plus(
                evalLeftSideCondition(operands.get(0), inputEvent),
                evalLeftSideCondition(operands.get(1), inputEvent));
          default:
            throw new UnsupportedOperationException(
                "the function is not supported for now: " + operator.getCepKind().toString());
        }
      } else {
        throw new IllegalStateException(
            "the left side CEP operation is not legal: " + inputOperation.getClass().toString());
      }
    }

    // evaluates the condition value (right) given an input event
    private CEPLiteral evalRightSideCondition(CEPOperation inputOperation, Event inputEvent) {
      if (inputOperation instanceof CEPLiteral) {
        return (CEPLiteral) inputOperation;
      } else if (inputOperation.getClass() == CEPCall.class) {
        CEPCall call = (CEPCall) inputOperation;
        CEPOperator operator = call.getOperator();
        List<CEPOperation> operands = call.getOperands();
        switch (operator.getCepKind()) {
          case PLUS:
            return plus(
                evalRightSideCondition(operands.get(0), inputEvent),
                evalRightSideCondition(operands.get(1), inputEvent));
          case PREV:
            return prev(operands.get(0), (CEPLiteral) operands.get(1), ptr, curEvent, inputEvent);
          default:
            throw new UnsupportedOperationException(
                "the function is not supported for now: " + operator.getCepKind().toString());
        }
      } else {
        throw new IllegalStateException(
            "the right side CEP operation is not legal: " + inputOperation.getClass().toString());
      }
    }

    /* below are function implementations */

    // represents the PREV operation
    private CEPLiteral prev(
        CEPOperation opr1,
        CEPLiteral opr2,
        EventPointer curPointer,
        Event curEvent,
        Event inputEvent) {
      if (opr1.getClass() != CEPFieldRef.class) {
        throw new IllegalStateException(
            "the first argument of the PREV operation should be a field reference. Provided: "
                + opr1.getClass().toString());
      }
      if (opr2.getTypeName() != Schema.TypeName.DECIMAL) {
        throw new IllegalStateException(
            "the second argument of the prev operation should be a decimal. Provided: "
                + opr2.getClass().toString());
      }
      // if the second argument is 0, return the current event
      if (opr2.getDecimal().intValue() == 0) {
        return inputEvent.toCEPLiteral((CEPFieldRef) opr1);
      }
      // for the starting event, return null
      if (curEvent == null) {
        return null;
      }
      CEPFieldRef fieldRef = (CEPFieldRef) opr1;
      String alpha = fieldRef.getAlpha(); // the patternVar
      int lastNumber = opr2.getDecimal().intValue();
      EventPointer iterPointer = curPointer.copy();

      while (curEvent != null && iterPointer.getPatternVar().equals(alpha) && lastNumber > 0) {
        iterPointer = curEvent.getPrevPointer(iterPointer);
        curEvent = curEvent.findEvent(iterPointer);
      }
      if (curEvent != null) {
        return curEvent.toCEPLiteral(fieldRef);
      } else {
        return null;
      }
    }

    // represents the LAST operation
    // TODO: add implementation. for now, return the current row
    private CEPLiteral last(CEPOperation opr1, CEPLiteral opr2, Event inputEvent) {
      if (opr1.getClass() != CEPFieldRef.class) {
        throw new IllegalStateException(
            "the first argument of the PREV operation should be a field reference. Provided: "
                + opr1.getClass().toString());
      }
      if (opr2.getTypeName() != Schema.TypeName.DECIMAL) {
        throw new IllegalStateException(
            "the second argument of the prev operation should be a decimal. Provided: "
                + opr2.getClass().toString());
      }
      return inputEvent.toCEPLiteral((CEPFieldRef) opr1);
    }

    // represents the PLUS operation
    private CEPLiteral plus(CEPLiteral opr1, CEPLiteral opr2) {
      Schema.TypeName type1 = opr1.getTypeName();
      Schema.TypeName type2 = opr2.getTypeName();
      if (type1.isNumericType() && type1 == type2) {
        switch (type1) {
          case BYTE:
            return CEPLiteral.of(opr1.getByte() + opr2.getByte());
          case INT16:
            return CEPLiteral.of(opr1.getInt16() + opr2.getInt16());
          case INT32:
            return CEPLiteral.of(opr1.getInt32() + opr2.getInt32());
          case INT64:
            return CEPLiteral.of(opr1.getInt64() + opr2.getInt64());
          case DECIMAL:
            return CEPLiteral.of(opr1.getDecimal().add(opr2.getDecimal()));
          case FLOAT:
            return CEPLiteral.of(opr1.getFloat() + opr2.getFloat());
          case DOUBLE:
            return CEPLiteral.of(opr1.getDouble() + opr2.getDouble());
          default:
            throw new UnsupportedOperationException("Type is not supported: " + type1.toString());
        }
      } else {
        throw new IllegalStateException(
            "Types do not match: " + type1.toString() + ", " + type2.toString());
      }
    }
  }

  // State are determined directly by the PATTERN clause
  private static class State implements Serializable {
    private int index = 0; // number ith state (pointer value) in the NFA
    private final String patternVar;
    private final Quantifier quant;
    private final CEPOperation
        condition; // condition to evaluate when taking the "begin" action and "proceed" action
    private State nextState = null;
    public final boolean isStart;
    public final boolean isFinal;
    private final boolean
        isKleenePlusSecondary; // whether is the second state for a Kleene Plus pattern variable

    State(
        String patternVar,
        Quantifier quant,
        CEPOperation condition,
        boolean isStart,
        boolean isFinal,
        boolean isKleenePlusSecondary) {
      this.patternVar = patternVar;
      this.quant = quant;
      this.condition = condition;
      this.isStart = isStart;
      this.isFinal = isFinal;
      this.isKleenePlusSecondary = isKleenePlusSecondary;
    }

    public String getPatternVar() {
      return patternVar;
    }

    // assigns a new index
    public int assignIndex() {
      return ++index;
    }

    // reset the index
    public void reset() {
      index = 0;
    }

    // checks if a state has a take edge
    public boolean hasTake() {
      return isKleenePlusSecondary;
    }

    // checks if a state has a proceed/begin edge
    public boolean hasProceed() {
      // except for the final state
      // return true
      return !isFinal;
    }

    public boolean isKleenePlusSecondary() {
      return isKleenePlusSecondary;
    }

    // get the condition for the `take` condition
    // returns null if the take edge takes no condition
    // throws exception if there is no take edge for the state
    public CEPOperation getTakeCondition() {
      if (!hasTake()) {
        throw new IllegalStateException("The state does not have a take edge.");
      }
      return condition;
    }

    // get the condition for the `proceed` or the `begin` edge
    // returns null if the proceed (begin) edge is always evaluated to true
    // throws exception if there is no proceed (begin) edge for the state
    public CEPOperation getProceedCondition() {
      if (!hasProceed()) {
        throw new IllegalStateException("The state does not have a proceed edge.");
      }
      // for singleton state or the first state of a paired state
      // return the current condition
      if (!isKleenePlusSecondary) {
        return condition;
      } else {
        return null;
      }
    }

    public Quantifier getQuantifier() {
      return quant;
    }

    public void setNextState(State theNextState) {

      if (this == theNextState) {
        return;
      }

      this.nextState = theNextState;
    }

    public State getNextState() {
      return nextState;
    }

    public boolean isKleenePlus() {
      return quant.toString().equals("+") || quant.toString().equals("+?");
    }
  }

  private static State setNextStatesAndAssignIndices(List<State> states) {
    for (int i = 0; i < (states.size() - 1); ++i) {
      State currentState = states.get(i);
      State nextState = states.get(i + 1);
      if (currentState.isKleenePlus()) {
        State secondaryState = currentState.getNextState();
        secondaryState.setNextState(nextState);
      } else {
        currentState.setNextState(nextState);
      }
    }
    return states.get(0);
  }

  // constructs states for the NFA and returns the start state
  private State loadStates(List<CEPPattern> patterns) {
    boolean startState;
    ArrayList<State> states = new ArrayList<>();

    for (int i = 0; i < patterns.size(); ++i) {
      if (i == 0) {
        startState = true;
      } else {
        startState = false;
      }

      CEPPattern currentPattern = patterns.get(i);
      CEPOperation condition = currentPattern.getPatternCondition();
      Quantifier quantifier = currentPattern.getQuantifier();

      if (quantifier.toString().equals("+")) {
        // for Kleene plus, we need a pair of states

        State primaryState =
            new State(
                currentPattern.getPatternVar(),
                Quantifier.PLUS,
                condition,
                startState,
                false,
                false);

        State secondaryState =
            new State(
                currentPattern.getPatternVar(),
                Quantifier.PLUS,
                condition,
                startState,
                false,
                true);

        primaryState.setNextState(secondaryState);
        states.add(primaryState);
      } else {
        // for non-Kleene-Plus pattern var, construct a single state
        State newState =
            new State(
                currentPattern.getPatternVar(), quantifier, condition, startState, false, false);
        states.add(newState);
      }
    }
    // add final state
    State theFinalState = new State("", Quantifier.NONE, null, false, true, false);
    states.add(theFinalState);
    State beginState = setNextStatesAndAssignIndices(states);
    return beginState;
  }
}