class BeamSqlSubstringExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, ImmutableMap<Integer, Object> correlateEnv) {
    String str = opValueEvaluated(0, inputRow, window, correlateEnv);
    int idx = opValueEvaluated(1, inputRow, window, correlateEnv);
    int startIdx = idx;
    if (startIdx > 0) {
      // NOTE: SQL substring is 1 based(rather than 0 based)
      startIdx -= 1;
    } else if (startIdx < 0) {
      // NOTE: SQL also support negative index...
      startIdx += str.length();
    } else {
      return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, "");
    }

    if (operands.size() == 3) {
      int length = opValueEvaluated(2, inputRow, window, correlateEnv);
      if (length < 0) {
        length = 0;
      }
      int endIdx = Math.min(startIdx + length, str.length());
      return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, str.substring(startIdx, endIdx));
    } else {
      return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, str.substring(startIdx));
    }
  }

  public  BeamSqlSubstringExpression(List<BeamSqlExpression> operands);

}