class BeamSqlOverlayExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, ImmutableMap<Integer, Object> correlateEnv) {
    String str = opValueEvaluated(0, inputRow, window, correlateEnv);
    String replaceStr = opValueEvaluated(1, inputRow, window, correlateEnv);
    int idx = opValueEvaluated(2, inputRow, window, correlateEnv);
    // the index is 1 based.
    idx -= 1;
    int length = replaceStr.length();
    if (operands.size() == 4) {
      length = opValueEvaluated(3, inputRow, window, correlateEnv);
    }

    StringBuilder result = new StringBuilder(str.length() + replaceStr.length() - length);
    result.append(str.substring(0, idx)).append(replaceStr).append(str.substring(idx + length));

    return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, result.toString());
  }

  public  BeamSqlOverlayExpression(List<BeamSqlExpression> operands);

}