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/08/03 17:05:49 UTC
[flink] branch master updated: [FLINK-9915] [table] Add TO_BASE64
function for table/sql API
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 e4c9d97 [FLINK-9915] [table] Add TO_BASE64 function for table/sql API
e4c9d97 is described below
commit e4c9d97a69555072c4940007754cd8b12a339d82
Author: yanghua <ya...@gmail.com>
AuthorDate: Mon Jul 23 17:52:02 2018 +0800
[FLINK-9915] [table] Add TO_BASE64 function for table/sql API
This closes #6390.
---
docs/dev/table/sql.md | 12 ++++++++++
docs/dev/table/tableApi.md | 26 ++++++++++++++++++++
.../flink/table/api/scala/expressionDsl.scala | 5 ++++
.../flink/table/codegen/calls/BuiltInMethods.scala | 2 ++
.../table/codegen/calls/FunctionGenerator.scala | 6 +++++
.../table/expressions/stringExpressions.scala | 27 +++++++++++++++++++++
.../table/functions/sql/ScalarSqlFunctions.scala | 9 +++++++
.../table/runtime/functions/ScalarFunctions.scala | 5 ++++
.../flink/table/validate/FunctionCatalog.scala | 2 ++
.../table/expressions/ScalarFunctionsTest.scala | 28 ++++++++++++++++++++++
10 files changed, 122 insertions(+)
diff --git a/docs/dev/table/sql.md b/docs/dev/table/sql.md
index 94fd59b..148f307 100644
--- a/docs/dev/table/sql.md
+++ b/docs/dev/table/sql.md
@@ -1859,6 +1859,18 @@ FROM_BASE64(text string)
<td>
<p>Returns the base string decoded with base64, if text is NULL, returns NULL. E.g. <code>FROM_BASE64('aGVsbG8gd29ybGQ=')</code> returns <code>hello world</code>.</p>
</td>
+ </tr>
+
+ <tr>
+ <td>
+ {% highlight text %}
+TO_BASE64(string)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns the base64-encoded result of <i>string</i>; returns NULL if <i>string</i> is NULL.</p>
+ <p>E.g., <code>TO_BASE64("hello world")</code> returns "aGVsbG8gd29ybGQ=".</p>
+ </td>
</tr>
</tbody>
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index 9317122..b702ddd 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -2496,6 +2496,19 @@ STRING.fromBase64()
<p>Returns the base string decoded with base64, if string is null, returns null. E.g. "aGVsbG8gd29ybGQ=".fromBase64() returns "hello world".</p>
</td>
</tr>
+
+ <tr>
+ <td>
+ {% highlight java %}
+STRING.toBase64()
+{% endhighlight %}
+ </td>
+
+ <td>
+ <p>Returns the base64-encoded result of <i>STRING</i>; retruns NULL if <i>STRING</i> is NULL.</p>
+ <p>E.g., <code>"hello world".toBase64()</code> returns "aGVsbG8gd29ybGQ=".</p>
+ </td>
+ </tr>
<tr>
<td>
@@ -4036,6 +4049,19 @@ STRING.initCap()
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight scala %}
+STRING.toBase64()
+{% endhighlight %}
+ </td>
+
+ <td>
+ <p>Returns the base64-encoded result of <i>STRING</i>; returns NULL if <i>STRING</i> is NULL.</p>
+ <p>E.g., <code>"hello world".toBase64()</code> returns "aGVsbG8gd29ybGQ=".</p>
+ </td>
+ </tr>
+
</tbody>
</table>
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 0989fcd..c7c805f 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
@@ -549,6 +549,11 @@ trait ImplicitExpressionOperations {
*/
def fromBase64() = FromBase64(expr)
+ /**
+ * Returns the base64-encoded result of the input string.
+ */
+ def toBase64() = ToBase64(expr)
+
// Temporal operations
/**
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 d08334f..f5ed9b3 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,4 +133,6 @@ object BuiltInMethods {
val BIN = Types.lookupMethod(classOf[JLong], "toBinaryString", classOf[Long])
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 230e3cf..74b69d6 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
@@ -152,6 +152,12 @@ object FunctionGenerator {
STRING_TYPE_INFO,
BuiltInMethods.FROMBASE64)
+ addSqlFunctionMethod(
+ TO_BASE64,
+ Seq(STRING_TYPE_INFO),
+ STRING_TYPE_INFO,
+ BuiltInMethods.TOBASE64)
+
// ----------------------------------------------------------------------------------------------
// Arithmetic functions
// ----------------------------------------------------------------------------------------------
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 87d251d..f844308 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
@@ -382,4 +382,31 @@ case class FromBase64(child: Expression) extends UnaryExpression with InputTypeS
}
override def toString: String = s"($child).fromBase64"
+
+}
+
+/**
+ * Returns the base64-encoded result of the input string.
+ */
+case class ToBase64(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"ToBase64 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.TO_BASE64, child.toRexNode)
+ }
+
+ override def toString: String = s"($child).toBase64"
+
}
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 49fda9d..21793e3 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
@@ -177,4 +177,13 @@ object ScalarSqlFunctions {
SqlFunctionCategory.STRING
)
+ val TO_BASE64 = new SqlFunction(
+ "TO_BASE64",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(ReturnTypes.explicit(SqlTypeName.VARCHAR), SqlTypeTransforms.TO_NULLABLE),
+ InferTypes.RETURN_TYPE,
+ OperandTypes.STRING,
+ SqlFunctionCategory.STRING
+ )
+
}
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 15f6096..03ee62c 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
@@ -207,4 +207,9 @@ object ScalarFunctions {
*/
def fromBase64(str: String): String = new String(Base64.decodeBase64(str))
+ /**
+ * Returns the base64-encoded result of the input string.
+ */
+ def toBase64(base: String): String = Base64.encodeBase64String(base.getBytes())
+
}
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 1e554ca..bca156c 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 {
"lpad" -> classOf[Lpad],
"rpad" -> classOf[Rpad],
"fromBase64" -> classOf[FromBase64],
+ "toBase64" -> classOf[ToBase64],
// math functions
"plus" -> classOf[Plus],
@@ -449,6 +450,7 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
ScalarSqlFunctions.SHA512,
ScalarSqlFunctions.SHA2,
ScalarSqlFunctions.FROM_BASE64,
+ ScalarSqlFunctions.TO_BASE64,
// EXTENSIONS
BasicOperatorTable.TUMBLE,
BasicOperatorTable.HOP,
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 d1ff806..6f9a9ae 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
@@ -472,6 +472,34 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"null")
}
+ @Test
+ def testToBase64(): Unit = {
+ testAllApis(
+ 'f0.toBase64(),
+ "f0.toBase64()",
+ "TO_BASE64(f0)",
+ "VGhpcyBpcyBhIHRlc3QgU3RyaW5nLg==")
+
+ testAllApis(
+ 'f8.toBase64(),
+ "f8.toBase64()",
+ "TO_BASE64(f8)",
+ "IFRoaXMgaXMgYSB0ZXN0IFN0cmluZy4g")
+
+ testAllApis(
+ "".toBase64(),
+ "''.toBase64()",
+ "TO_BASE64('')",
+ "")
+
+ //null test
+ testAllApis(
+ 'f33.toBase64(),
+ "f33.toBase64()",
+ "TO_BASE64(f33)",
+ "null")
+ }
+
// ----------------------------------------------------------------------------------------------
// Math functions
// ----------------------------------------------------------------------------------------------