class BeamSqlTrimExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, ImmutableMap<Integer, Object> correlateEnv) {
    if (operands.size() == 1) {
      return BeamSqlPrimitive.of(
          SqlTypeName.VARCHAR,
          opValueEvaluated(0, inputRow, window, correlateEnv).toString().trim());
    } else {
      SqlTrimFunction.Flag type = opValueEvaluated(0, inputRow, window, correlateEnv);
      String targetStr = opValueEvaluated(1, inputRow, window, correlateEnv);
      String containingStr = opValueEvaluated(2, inputRow, window, correlateEnv);

      switch (type) {
        case LEADING:
          return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, leadingTrim(containingStr, targetStr));
        case TRAILING:
          return BeamSqlPrimitive.of(SqlTypeName.VARCHAR, trailingTrim(containingStr, targetStr));
        case BOTH:
        default:
          return BeamSqlPrimitive.of(
              SqlTypeName.VARCHAR, trailingTrim(leadingTrim(containingStr, targetStr), targetStr));
      }
    }
  }

  public  BeamSqlTrimExpression(List<BeamSqlExpression> operands);

}