You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2018/09/15 07:29:58 UTC
[GitHub] asfgit closed pull request #6450: [FLINK-9991] [table] Add
regexp_replace supported in TableAPI and SQL
asfgit closed pull request #6450: [FLINK-9991] [table] Add regexp_replace supported in TableAPI and SQL
URL: https://github.com/apache/flink/pull/6450
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/dev/table/functions.md b/docs/dev/table/functions.md
index e41fe4561b9..fe8e6e1c0f6 100644
--- a/docs/dev/table/functions.md
+++ b/docs/dev/table/functions.md
@@ -2413,6 +2413,18 @@ REPEAT(string, integer)
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight text %}
+REGEXP_REPLACE(string1, string2, string3)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>string1</i> with all the substrings that match a regular expression <i>string2</i> being replaced with <i>string3</i>.</p>
+ <p>E.g. <code>REGEXP_REPLACE('foobar', 'oo|ar', '')</code> returns "fb".</p>
+ </td>
+ </tr>
+
<tr>
<td>
{% highlight text %}
@@ -2638,6 +2650,18 @@ STRING.repeat(INT)
<p>E.g., <code>'This is a test String.'.repeat(2)</code> returns "This is a test String.This is a test String.".</p>
</td>
</tr>
+
+ <tr>
+ <td>
+ {% highlight java %}
+STRING1.regexpReplace(STRING2, STRING3)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>STRING1</i> with all the substrings that match a regular expression <i>STRING2</i> being replaced with <i>STRING3</i>.</p>
+ <p>E.g., <code>'foobar'.regexpReplace('oo|ar', '')</code> returns "fb".</p>
+ </td>
+ </tr>
<tr>
<td>
@@ -2866,6 +2890,18 @@ STRING.repeat(INT)
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight scala %}
+STRING1.regexpReplace(STRING2, STRING3)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>STRING1</i> with all the substrings that match a regular expression <i>STRING2</i> being replaced with <i>STRING3</i>.</p>
+ <p>E.g. <code>"foobar".regexpReplace("oo|ar", "")</code> returns "fb".</p>
+ </td>
+ </tr>
+
<tr>
<td>
{% highlight scala %}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
index cf8cd91bb2f..e5c446cc3e7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
@@ -564,6 +564,12 @@ trait ImplicitExpressionOperations {
def overlay(newString: Expression, starting: Expression, length: Expression) =
Overlay(expr, newString, starting, length)
+ /**
+ * Returns a string with all substrings that match the regular expression being replaced.
+ */
+ def regexpReplace(regex: Expression, replacement: Expression) =
+ RegexpReplace(expr, regex, replacement)
+
/**
* Returns the base string decoded with base64.
*/
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
index 899cb0ff35a..f15da166725 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
@@ -133,6 +133,13 @@ object BuiltInMethods {
val BIN = Types.lookupMethod(classOf[JLong], "toBinaryString", classOf[Long])
+ val REGEXP_REPLACE = Types.lookupMethod(
+ classOf[ScalarFunctions],
+ "regexp_replace",
+ classOf[String],
+ classOf[String],
+ classOf[String])
+
val FROMBASE64 = Types.lookupMethod(classOf[ScalarFunctions], "fromBase64", classOf[String])
val TOBASE64 = Types.lookupMethod(classOf[ScalarFunctions], "toBase64", classOf[String])
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
index c7eb869a5b5..b6eb3e8fe00 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
@@ -146,6 +146,12 @@ object FunctionGenerator {
STRING_TYPE_INFO,
BuiltInMethod.OVERLAY.method)
+ addSqlFunctionMethod(
+ REGEXP_REPLACE,
+ Seq(STRING_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO),
+ STRING_TYPE_INFO,
+ BuiltInMethods.REGEXP_REPLACE)
+
addSqlFunctionMethod(
FROM_BASE64,
Seq(STRING_TYPE_INFO),
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/stringExpressions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/stringExpressions.scala
index 70794407cc0..3003b8cebf5 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/stringExpressions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/stringExpressions.scala
@@ -358,6 +358,26 @@ case class Rpad(text: Expression, len: Expression, pad: Expression)
}
}
+/**
+ * Returns a string with all substrings that match the regular expression being replaced.
+ */
+case class RegexpReplace(str: Expression, regex: Expression, replacement: Expression)
+ extends Expression with InputTypeSpec {
+
+ override private[flink] def resultType: TypeInformation[_] = BasicTypeInfo.STRING_TYPE_INFO
+
+ override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
+ Seq(STRING_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO)
+
+ override private[flink] def children: Seq[Expression] = Seq(str, regex, replacement)
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ relBuilder.call(ScalarSqlFunctions.REGEXP_REPLACE, children.map(_.toRexNode))
+ }
+
+ override def toString: String = s"($str).regexp_replace($regex, $replacement)"
+}
+
/**
* Returns the base string decoded with base64.
* Returns NULL If the input string is NULL.
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
index db67e3985db..249eb8a8840 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
@@ -187,6 +187,16 @@ object ScalarSqlFunctions {
SqlFunctionCategory.TIMEDATE
)
+ val REGEXP_REPLACE = new SqlFunction(
+ "REGEXP_REPLACE",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(
+ ReturnTypes.explicit(SqlTypeName.VARCHAR), SqlTypeTransforms.TO_NULLABLE),
+ InferTypes.RETURN_TYPE,
+ OperandTypes.STRING_STRING_STRING,
+ SqlFunctionCategory.STRING
+ )
+
val FROM_BASE64 = new SqlFunction(
"FROM_BASE64",
SqlKind.OTHER_FUNCTION,
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
index fa6f020ea6d..35aad39e30b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.functions
import java.lang.{StringBuilder, Long => JLong}
import java.math.{BigDecimal => JBigDecimal}
import java.nio.charset.StandardCharsets
+import java.util.regex.Matcher
import org.apache.commons.codec.binary.{Base64, Hex}
import org.apache.commons.lang3.StringUtils
@@ -205,6 +206,18 @@ object ScalarFunctions {
new String(data)
}
+
+ /**
+ * Returns a string with all substrings that match the regular expression being replaced.
+ */
+ def regexp_replace(str: String, regex: String, replacement: String): String = {
+ if (str == null || regex == null || replacement == null) {
+ return null
+ }
+
+ str.replaceAll(regex, Matcher.quoteReplacement(replacement))
+ }
+
/**
* Returns the base string decoded with base64.
*/
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
index fe508aa1f4a..adb4468a82e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
@@ -202,6 +202,7 @@ object FunctionCatalog {
"concat_ws" -> classOf[ConcatWs],
"lpad" -> classOf[Lpad],
"rpad" -> classOf[Rpad],
+ "regexpReplace" -> classOf[RegexpReplace],
"fromBase64" -> classOf[FromBase64],
"toBase64" -> classOf[ToBase64],
"uuid" -> classOf[UUID],
@@ -455,6 +456,7 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
ScalarSqlFunctions.SHA384,
ScalarSqlFunctions.SHA512,
ScalarSqlFunctions.SHA2,
+ ScalarSqlFunctions.REGEXP_REPLACE,
ScalarSqlFunctions.FROM_BASE64,
ScalarSqlFunctions.TO_BASE64,
ScalarSqlFunctions.UUID,
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
index fbd9b028547..c826c62a646 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
@@ -550,6 +550,99 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"1111111111111111111111111111111111111111111111111111111111111111")
}
+ @Test
+ def testRegexpReplace(): Unit = {
+ testAllApis(
+ "foobar".regexpReplace("oo|ar", ""),
+ "'foobar'.regexpReplace('oo|ar', '')",
+ "regexp_replace('foobar', 'oo|ar', '')",
+ "fb")
+
+ testAllApis(
+ "foobar".regexpReplace("oo|ar", "abc"),
+ "'foobar'.regexpReplace('oo|ar', 'abc')",
+ "regexp_replace('foobar', 'oo|ar', 'abc')",
+ "fabcbabc")
+
+ testAllApis(
+ "fooor".regexpReplace("oo", ""),
+ "'fooor'.regexpReplace('oo', '')",
+ "REGEXP_REPLACE('fooor', 'oo', '')",
+ "for")
+
+ testAllApis(
+ 'f33.regexpReplace("oo|ar", ""),
+ "f33.regexpReplace('oo|ar', '')",
+ "REGEXP_REPLACE(f33, 'oo|ar', '')",
+ "null")
+
+ testAllApis(
+ "foobar".regexpReplace('f33, ""),
+ "'foobar'.regexpReplace(f33, '')",
+ "REGEXP_REPLACE('foobar', f33, '')",
+ "null")
+
+ testAllApis(
+ "foobar".regexpReplace("oo|ar", 'f33),
+ "'foobar'.regexpReplace('oo|ar', f33)",
+ "REGEXP_REPLACE('foobar', 'oo|ar', f33)",
+ "null")
+
+ testAllApis(
+ "foobar".regexpReplace("^f", ""),
+ "'foobar'.regexpReplace('^f', '')",
+ "regexp_replace('foobar', '^f', '')",
+ "oobar")
+
+ testAllApis(
+ "foobar".regexpReplace("r$", ""),
+ "'foobar'.regexpReplace('r$', '')",
+ "regexp_replace('foobar', 'r$', '')",
+ "fooba")
+
+ testAllApis(
+ "foobar".regexpReplace("^f*.*r$", ""),
+ "'foobar'.regexpReplace('^f*.*r$', '')",
+ "regexp_replace('foobar', '^f*.*r$', '')",
+ "")
+
+ testAllApis(
+ "foobar".regexpReplace("\\d", ""),
+ "'foobar'.regexpReplace('\\d', '')",
+ "regexp_replace('foobar', '\\d', '')",
+ "foobar")
+
+ testAllApis(
+ "foobar".regexpReplace("\\w", ""),
+ "'foobar'.regexpReplace('\\w', '')",
+ "regexp_replace('foobar', '\\w', '')",
+ "")
+
+ testAllApis(
+ "foobar".regexpReplace("$b", ""),
+ "'foobar'.regexpReplace('$b', '')",
+ "regexp_replace('foobar', '$b', '')",
+ "foobar")
+
+ testAllApis(
+ "foobar".regexpReplace("/b", ""),
+ "'foobar'.regexpReplace('/b', '')",
+ "regexp_replace('foobar', '/b', '')",
+ "foobar")
+
+ testAllApis(
+ "foobar".regexpReplace("oo", "$"),
+ "'foobar'.regexpReplace('oo', '$')",
+ "regexp_replace('foobar', 'oo', '$')",
+ "f$bar")
+
+ testAllApis(
+ "foobar".regexpReplace("oo", "\\"),
+ "'foobar'.regexpReplace('oo', '\\')",
+ "regexp_replace('foobar', 'oo', '\\')",
+ "f\\bar")
+ }
+
@Test
def testFromBase64(): Unit = {
testAllApis(
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
index 4a79a61d977..b49922bc3fc 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
@@ -154,6 +154,7 @@ class SqlExpressionTest extends ExpressionTestBase {
testSqlApi(
"REPEAT('This is a test String.', 2)",
"This is a test String.This is a test String.")
+ testSqlApi("REGEXP_REPLACE('foobar', 'oo|ar', '')", "fb")
}
@Test
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services