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/08/03 17:07:09 UTC
[GitHub] asfgit closed pull request #6390: [FLINK-9915] [table] Add
TO_BASE64 function for table/sql API
asfgit closed pull request #6390: [FLINK-9915] [table] Add TO_BASE64 function for table/sql API
URL: https://github.com/apache/flink/pull/6390
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/sql.md b/docs/dev/table/sql.md
index 366e3fdcc64..4d1b27b018d 100644
--- a/docs/dev/table/sql.md
+++ b/docs/dev/table/sql.md
@@ -1848,6 +1848,17 @@ 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 the input string. If the input is NULL, returns NULL. E.g. <code>TO_BASE64('hello world')</code> returns <code>aGVsbG8gd29ybGQ=</code>.</p>
+ </td>
</tr>
</tbody>
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index 6e202f19d5d..367f6cf6e15 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -2474,6 +2474,7 @@ STRING.rpad(len INT, pad STRING)
<p>Returns a string right-padded with the given pad string to a length of len characters. If the string is longer than len, the return value is shortened to len characters. E.g. "hi".rpad(4, '??') returns "hi??", "hi".rpad(1, '??') returns "h".</p>
</td>
</tr>
+
<tr>
<td>
{% highlight java %}
@@ -2485,6 +2486,18 @@ 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 STRING. If STRING is NULL, returns NULL. E.g. "hello world".toBase64() returns "aGVsbG8gd29ybGQ=".</p>
+ </td>
+ </tr>
<tr>
<td>
@@ -4025,6 +4038,18 @@ STRING.initCap()
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight scala %}
+STRING.toBase64()
+{% endhighlight %}
+ </td>
+
+ <td>
+ <p>Returns the base64-encoded result of STRING. If STRING is NULL, returns NULL. E.g. "hello world".toBase64() 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 35d2167848a..9e8acb1469e 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 0e0f709eabc..512ae7e8e28 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
@@ -114,4 +114,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 a5c275ab415..57d32c6e6e1 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 87d251deb75..6a3203ac998 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, children.map(_.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 49fda9dd46d..21793e3351c 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 50e8f9cc900..9417514fb89 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
@@ -201,4 +201,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 8a91340d86f..3668d4dd903 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],
@@ -447,6 +448,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 4eb2e33a8a0..6f9a45664e7 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
// ----------------------------------------------------------------------------------------------
----------------------------------------------------------------
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