You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xc...@apache.org on 2018/10/09 21:33:51 UTC
[flink] branch master updated: [FLINK-9990] [table] Add
regex_extract function in TableAPI and SQL
This is an automated email from the ASF dual-hosted git repository.
xccui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5dc3609 [FLINK-9990] [table] Add regex_extract function in TableAPI and SQL
5dc3609 is described below
commit 5dc360984143005f73b8f70f97ed6b1c2afd7dc3
Author: yanghua <ya...@gmail.com>
AuthorDate: Sun Jul 29 11:27:51 2018 +0800
[FLINK-9990] [table] Add regex_extract function in TableAPI and SQL
This closes #6448.
---
docs/dev/table/functions.md | 39 ++++++++++++++++
.../flink/table/api/scala/expressionDsl.scala | 12 +++++
.../flink/table/codegen/calls/BuiltInMethods.scala | 13 ++++++
.../table/codegen/calls/FunctionGenerator.scala | 12 +++++
.../table/expressions/stringExpressions.scala | 39 ++++++++++++++++
.../table/functions/sql/ScalarSqlFunctions.scala | 12 +++++
.../table/runtime/functions/ScalarFunctions.scala | 26 +++++++++++
.../flink/table/validate/FunctionCatalog.scala | 2 +
.../table/expressions/ScalarFunctionsTest.scala | 52 ++++++++++++++++++++++
.../table/expressions/SqlExpressionTest.scala | 1 +
10 files changed, 208 insertions(+)
diff --git a/docs/dev/table/functions.md b/docs/dev/table/functions.md
index f888d79..8a22558 100644
--- a/docs/dev/table/functions.md
+++ b/docs/dev/table/functions.md
@@ -2499,6 +2499,19 @@ REPLACE(string1, string2, string3)
<tr>
<td>
{% highlight text %}
+REGEXP_EXTRACT(string1, string2[, integer])
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>string1</i> which extracted with a specified regular expression <i>string2</i> and a regex match group index <i>integer</i>.</p>
+ <p><b>Note:</b> The regex match group index starts from 1 and 0 means matching the whole regex. In addition, the regex match group index should not exceed the number of the defined groups.</p>
+ <p>E.g. <code>REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)"</code> returns "bar".</p>
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ {% highlight text %}
INITCAP(string)
{% endhighlight %}
</td>
@@ -2751,6 +2764,19 @@ STRING1.replace(STRING2, STRING3)
<tr>
<td>
{% highlight java %}
+STRING1.regexpExtract(STRING2[, INTEGER1])
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>STRING1</i> which extracted with a specified regular expression <i>STRING2</i> and a regex match group index <i>INTEGER1</i>.</p>
+ <p><b>Note:</b> The regex match group index starts from 1 and 0 means matching the whole regex. In addition, the regex match group index should not exceed the number of the defined groups.</p>
+ <p>E.g. <code>'foothebar'.regexpExtract('foo(.*?)(bar)', 2)"</code> returns "bar".</p>
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ {% highlight java %}
STRING.initCap()
{% endhighlight %}
</td>
@@ -3002,6 +3028,19 @@ STRING1.replace(STRING2, STRING3)
<tr>
<td>
{% highlight scala %}
+STRING1.regexpExtract(STRING2[, INTEGER1])
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns a string from <i>STRING1</i> which extracted with a specified regular expression <i>STRING2</i> and a regex match group index <i>INTEGER1</i>.</p>
+ <p><b>Note:</b> The regex match group index starts from 1 and 0 means matching the whole regex. In addition, the regex match group index should not exceed the number of the defined groups.</p>
+ <p>E.g. <code>"foothebar".regexpExtract("foo(.*?)(bar)", 2)"</code> returns "bar".</p>
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ {% highlight scala %}
STRING.initCap()
{% endhighlight %}
</td>
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 e638051..e1947c3 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
@@ -585,6 +585,18 @@ trait ImplicitExpressionOperations {
RegexpReplace(expr, regex, replacement)
/**
+ * Returns a string extracted with a specified regular expression and a regex match group index.
+ */
+ def regexpExtract(regex: Expression, extractIndex: Expression) =
+ RegexpExtract(expr, regex, extractIndex)
+
+ /**
+ * Returns a string extracted with a specified regular expression.
+ */
+ def regexpExtract(regex: Expression) =
+ RegexpExtract(expr, regex, null)
+
+ /**
* Returns the base string decoded with base64.
*/
def fromBase64() = FromBase64(expr)
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 34be86e..7781b57 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
@@ -143,6 +143,19 @@ object BuiltInMethods {
classOf[String],
classOf[String])
+ val REGEXP_EXTRACT = Types.lookupMethod(
+ classOf[ScalarFunctions],
+ "regexp_extract",
+ classOf[String],
+ classOf[String],
+ classOf[Integer])
+
+ val REGEXP_EXTRACT_WITHOUT_INDEX = Types.lookupMethod(
+ classOf[ScalarFunctions],
+ "regexp_extract",
+ 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 58c476a..6404813 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
@@ -160,6 +160,18 @@ object FunctionGenerator {
BuiltInMethod.REPLACE.method)
addSqlFunctionMethod(
+ REGEXP_EXTRACT,
+ Seq(STRING_TYPE_INFO, STRING_TYPE_INFO, INT_TYPE_INFO),
+ STRING_TYPE_INFO,
+ BuiltInMethods.REGEXP_EXTRACT)
+
+ addSqlFunctionMethod(
+ REGEXP_EXTRACT,
+ Seq(STRING_TYPE_INFO, STRING_TYPE_INFO),
+ STRING_TYPE_INFO,
+ BuiltInMethods.REGEXP_EXTRACT_WITHOUT_INDEX)
+
+ addSqlFunctionMethod(
FROM_BASE64,
Seq(STRING_TYPE_INFO),
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 d0ae246..d65de86 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
@@ -383,6 +383,45 @@ case class RegexpReplace(str: Expression, regex: Expression, replacement: Expres
}
/**
+ * Returns a string extracted with a specified regular expression and a regex match group index.
+ */
+case class RegexpExtract(str: Expression, regex: Expression, extractIndex: Expression)
+ extends Expression with InputTypeSpec {
+ def this(str: Expression, regex: Expression) = this(str, regex, null)
+
+ override private[flink] def resultType: TypeInformation[_] = BasicTypeInfo.STRING_TYPE_INFO
+
+ override private[flink] def expectedTypes: Seq[TypeInformation[_]] = {
+ if (extractIndex == null) {
+ Seq(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
+ } else {
+ Seq(
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO)
+ }
+ }
+
+ override private[flink] def children: Seq[Expression] = {
+ if (extractIndex == null) {
+ Seq(str, regex)
+ } else {
+ Seq(str, regex, extractIndex)
+ }
+ }
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ relBuilder.call(ScalarSqlFunctions.REGEXP_EXTRACT, children.map(_.toRexNode))
+ }
+
+ override def toString: String = s"($str).regexp_extract($regex, $extractIndex)"
+}
+
+object RegexpExtract {
+ def apply(str: Expression, regex: Expression): RegexpExtract = RegexpExtract(str, regex, null)
+}
+
+/**
* 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 c7b68d0..0f594bb 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
@@ -205,6 +205,18 @@ object ScalarSqlFunctions {
SqlFunctionCategory.STRING
)
+ val REGEXP_EXTRACT = new SqlFunction(
+ "REGEXP_EXTRACT",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(
+ ReturnTypes.explicit(SqlTypeName.VARCHAR), SqlTypeTransforms.FORCE_NULLABLE),
+ InferTypes.RETURN_TYPE,
+ OperandTypes.or(OperandTypes.STRING_STRING_INTEGER,
+ OperandTypes.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 0c46992..1db4da2 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
@@ -21,6 +21,7 @@ import java.lang.{StringBuilder, Long => JLong}
import java.math.{BigDecimal => JBigDecimal}
import java.nio.charset.StandardCharsets
import java.util.regex.Matcher
+import java.util.regex.Pattern
import org.apache.commons.codec.binary.{Base64, Hex}
import org.apache.commons.lang3.StringUtils
@@ -227,6 +228,31 @@ object ScalarFunctions {
}
/**
+ * Returns a string extracted with a specified regular expression and a regex match group index.
+ */
+ def regexp_extract(str: String, regex: String, extractIndex: Integer): String = {
+ if (str == null || regex == null) {
+ return null
+ }
+
+ val m = Pattern.compile(regex).matcher(str)
+ if (m.find) {
+ val mr = m.toMatchResult
+ return mr.group(extractIndex)
+ }
+
+ null
+ }
+
+ /**
+ * Returns a string extracted with a specified regular expression and
+ * a optional regex match group index.
+ */
+ def regexp_extract(str: String, regex: String): String = {
+ regexp_extract(str, regex, 0)
+ }
+
+ /**
* Returns the base string decoded with base64.
*/
def fromBase64(str: String): String =
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 218ffec..26a20ff 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
@@ -203,6 +203,7 @@ object FunctionCatalog {
"concat_ws" -> classOf[ConcatWs],
"lpad" -> classOf[Lpad],
"rpad" -> classOf[Rpad],
+ "regexpExtract" -> classOf[RegexpExtract],
"fromBase64" -> classOf[FromBase64],
"toBase64" -> classOf[ToBase64],
"uuid" -> classOf[UUID],
@@ -462,6 +463,7 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
ScalarSqlFunctions.SHA384,
ScalarSqlFunctions.SHA512,
ScalarSqlFunctions.SHA2,
+ ScalarSqlFunctions.REGEXP_EXTRACT,
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 541ad62..6422dd1 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
@@ -648,6 +648,58 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
}
@Test
+ def testRegexpExtract(): Unit = {
+ testAllApis(
+ "foothebar".regexpExtract("foo(.*?)(bar)", 2),
+ "'foothebar'.regexpExtract('foo(.*?)(bar)', 2)",
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)",
+ "bar")
+
+ testAllApis(
+ "foothebar".regexpExtract("foo(.*?)(bar)", 0),
+ "'foothebar'.regexpExtract('foo(.*?)(bar)', 0)",
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 0)",
+ "foothebar")
+
+ testAllApis(
+ "foothebar".regexpExtract("foo(.*?)(bar)", 1),
+ "'foothebar'.regexpExtract('foo(.*?)(bar)', 1)",
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 1)",
+ "the")
+
+ testAllApis(
+ "foothebar".regexpExtract("foo([\\w]+)", 1),
+ "'foothebar'.regexpExtract('foo([\\w]+)', 1)",
+ "REGEXP_EXTRACT('foothebar', 'foo([\\w]+)', 1)",
+ "thebar")
+
+ testAllApis(
+ "foothebar".regexpExtract("foo([\\d]+)", 1),
+ "'foothebar'.regexpExtract('foo([\\d]+)', 1)",
+ "REGEXP_EXTRACT('foothebar', 'foo([\\d]+)', 1)",
+ "null")
+
+ testAllApis(
+ 'f33.regexpExtract("foo(.*?)(bar)", 2),
+ "f33.regexpExtract('foo(.*?)(bar)', 2)",
+ "REGEXP_EXTRACT(f33, 'foo(.*?)(bar)', 2)",
+ "null")
+
+ testAllApis(
+ "foothebar".regexpExtract('f33, 2),
+ "'foothebar'.regexpExtract(f33, 2)",
+ "REGEXP_EXTRACT('foothebar', f33, 2)",
+ "null")
+
+ //test for optional group index
+ testAllApis(
+ "foothebar".regexpExtract("foo(.*?)(bar)"),
+ "'foothebar'.regexpExtract('foo(.*?)(bar)')",
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)')",
+ "foothebar")
+ }
+
+ @Test
def testFromBase64(): Unit = {
testAllApis(
'f35.fromBase64(),
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 8949a2f..93285d0 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
@@ -157,6 +157,7 @@ class SqlExpressionTest extends ExpressionTestBase {
"This is a test String.This is a test String.")
testSqlApi("REGEXP_REPLACE('foobar', 'oo|ar', '')", "fb")
testSqlApi("REPLACE('hello world', 'world', 'flink')", "hello flink")
+ testSqlApi("REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)", "bar")
}
@Test