[FLINK-10060] [table] Add RTRIM function in Table API and SQL

This closes apache#6509.

docs/dev/table/functions.md

@@ -2389,6 +2389,18 @@ LTRIM(string)
  </td>
  </tr>
 
+ <tr>
+  <td>
+  {% highlight text %}
+RTRIM(string)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string that removes the right whitespaces from <i>string</i>.</p> 
+ <p>E.g., <code>RTRIM('This is a test String. ')</code> returns "This is a test String.".</p>
+ </td>
+ </tr>
+
  <tr>
   <td>
   {% highlight text %}
@@ -2590,6 +2602,18 @@ STRING.ltrim()
  <p>E.g., <code>' This is a test String.'.ltrim()</code> returns "This is a test String.".</p>
  </td>
  </tr>
+
+ <tr>
+  <td>
+  {% highlight java %}
+STRING.rtrim()
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string that removes the right whitespaces from <i>STRING</i>.</p> 
+ <p>E.g., <code>'This is a test String. '.rtrim()</code> returns "This is a test String.".</p>
+ </td>
+ </tr>
 
  <tr>
  <td>
@@ -2794,6 +2818,18 @@ STRING.ltrim()
  </td>
  </tr>
 
+ <tr>
+  <td>
+  {% highlight scala %}
+STRING.rtrim()
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string that removes the right whitespaces from <i>STRING</i>.</p> 
+ <p>E.g., <code>"This is a test String. ".rtrim()</code> returns "This is a test String.".</p>
+ </td>
+ </tr>
+
  <tr>
   <td>
   {% highlight scala %}

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala

@@ -568,6 +568,11 @@ trait ImplicitExpressionOperations {
  */
  def ltrim() = LTrim(expr)
 
+ /**
+ * Returns a string that removes the right whitespaces from the given string.
+ */
+ def rtrim() = RTrim(expr)
+
  // Temporal operations
 
  /**

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala

@@ -170,6 +170,12 @@ object FunctionGenerator {
  STRING_TYPE_INFO,
  BuiltInMethod.LTRIM.method)
 
+ addSqlFunctionMethod(
+ RTRIM,
+ Seq(STRING_TYPE_INFO),
+ STRING_TYPE_INFO,
+ BuiltInMethod.RTRIM.method)
+
  // ----------------------------------------------------------------------------------------------
  // Arithmetic functions
  // ----------------------------------------------------------------------------------------------

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/stringExpressions.scala

@@ -434,3 +434,28 @@ case class LTrim(child: Expression) extends UnaryExpression with InputTypeSpec {
 
  override def toString = s"($child).ltrim"
 }
+
+/**
+ * Returns a string that removes the right whitespaces from the given string.
+ */
+case class RTrim(child: Expression) extends UnaryExpression with InputTypeSpec {
+
+ override private[flink] def expectedTypes: Seq[TypeInformation[_]] = Seq(STRING_TYPE_INFO)
+
+ override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
+
+ override private[flink] def validateInput(): ValidationResult = {
+ if (child.resultType == STRING_TYPE_INFO) {
+ ValidationSuccess
+ } else {
+ ValidationFailure(s"RTrim operator requires a String input, " +
+ s"but $child is of type ${child.resultType}")
+ }
+ }
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ relBuilder.call(ScalarSqlFunctions.RTRIM, child.toRexNode)
+ }
+
+ override def toString = s"($child).rtrim"
+}

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala

@@ -214,4 +214,12 @@ object ScalarSqlFunctions {
  OperandTypes.STRING,
  SqlFunctionCategory.STRING)
 
+ val RTRIM = new SqlFunction(
+ "RTRIM",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(ReturnTypes.explicit(SqlTypeName.VARCHAR), SqlTypeTransforms.TO_NULLABLE),
+ InferTypes.RETURN_TYPE,
+ OperandTypes.STRING,
+ SqlFunctionCategory.STRING)
+
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala

@@ -206,6 +206,7 @@ object FunctionCatalog {
  "toBase64" -> classOf[ToBase64],
  "uuid" -> classOf[UUID],
  "ltrim" -> classOf[LTrim],
+ "rtrim" -> classOf[RTrim],
 
  // math functions
  "plus" -> classOf[Plus],
@@ -457,6 +458,7 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
  ScalarSqlFunctions.TO_BASE64,
  ScalarSqlFunctions.UUID,
  ScalarSqlFunctions.LTRIM,
+ ScalarSqlFunctions.RTRIM,
 
  // EXTENSIONS
  BasicOperatorTable.TUMBLE,

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala

@@ -653,6 +653,33 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  "null")
  }
 
+ @Test
+ def testRTrim(): Unit = {
+ testAllApis(
+ 'f8.rtrim(),
+ "f8.rtrim",
+ "RTRIM(f8)",
+ " This is a test String.")
+
+ testAllApis(
+ 'f0.rtrim(),
+ "f0.rtrim",
+ "RTRIM(f0)",
+ "This is a test String.")
+
+ testAllApis(
+ "".rtrim(),
+ "''.rtrim()",
+ "RTRIM('')",
+ "")
+
+ testAllApis(
+ 'f33.rtrim(),
+ "f33.rtrim",
+ "RTRIM(f33)",
+ "null")
+ }
+
  // ----------------------------------------------------------------------------------------------
  // Math functions
  // ----------------------------------------------------------------------------------------------

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala

@@ -137,6 +137,7 @@ class SqlExpressionTest extends ExpressionTestBase {
  testSqlApi("TRIM(LEADING 'x' FROM 'xxxxSTRINGxxxx')", "STRINGxxxx")
  testSqlApi("TRIM(TRAILING 'x' FROM 'xxxxSTRINGxxxx')", "xxxxSTRING")
  testSqlApi("LTRIM(' This is a test String.')", "This is a test String.")
+ testSqlApi("RTRIM('This is a test String. ')", "This is a test String.")
  testSqlApi(
  "OVERLAY('This is an old string' PLACING ' new' FROM 10 FOR 5)",
  "This is a new string")