You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/09/02 08:41:53 UTC
[flink] 04/05: [hotfix][table-planner] Ensure that IS JSON never
returns NULL
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 74146c626edd7603f00a6f0c90c7fccaf2066912
Author: Ingo Bürk <in...@tngtech.com>
AuthorDate: Wed Sep 1 10:00:25 2021 +0200
[hotfix][table-planner] Ensure that IS JSON never returns NULL
---
.../planner/codegen/calls/FunctionGenerator.scala | 67 ++++++++++-----------
.../planner/codegen/calls/MethodCallGen.scala | 70 ++++++++++++----------
2 files changed, 69 insertions(+), 68 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
index 04dcc5f..71ed038 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
@@ -789,53 +789,45 @@ class FunctionGenerator private(config: TableConfig) {
addSqlFunctionMethod(JSON_EXISTS, Seq(CHAR, CHAR, RAW), BuiltInMethods.JSON_EXISTS_ON_ERROR)
addSqlFunctionMethod(JSON_EXISTS, Seq(VARCHAR, CHAR, RAW), BuiltInMethods.JSON_EXISTS_ON_ERROR)
- addSqlFunctionMethod(IS_JSON_VALUE, Seq(CHAR), BuiltInMethod.IS_JSON_VALUE.method)
- addSqlFunctionMethod(IS_JSON_VALUE, Seq(VARCHAR), BuiltInMethod.IS_JSON_VALUE.method)
-
- addSqlFunctionMethod(IS_JSON_OBJECT, Seq(CHAR), BuiltInMethod.IS_JSON_OBJECT.method)
- addSqlFunctionMethod(IS_JSON_OBJECT, Seq(VARCHAR), BuiltInMethod.IS_JSON_OBJECT.method)
-
- addSqlFunctionMethod(IS_JSON_ARRAY, Seq(CHAR), BuiltInMethod.IS_JSON_ARRAY.method)
- addSqlFunctionMethod(IS_JSON_ARRAY, Seq(VARCHAR), BuiltInMethod.IS_JSON_ARRAY.method)
-
- addSqlFunctionMethod(IS_JSON_SCALAR, Seq(CHAR), BuiltInMethod.IS_JSON_SCALAR.method)
- addSqlFunctionMethod(IS_JSON_SCALAR, Seq(VARCHAR), BuiltInMethod.IS_JSON_SCALAR.method)
+ addSqlFunctionMethod(IS_JSON_VALUE, Seq(CHAR),
+ BuiltInMethod.IS_JSON_VALUE.method, argsNullable = true)
+ addSqlFunctionMethod(IS_JSON_VALUE, Seq(VARCHAR),
+ BuiltInMethod.IS_JSON_VALUE.method, argsNullable = true)
+
+ addSqlFunctionMethod(IS_JSON_OBJECT, Seq(CHAR),
+ BuiltInMethod.IS_JSON_OBJECT.method, argsNullable = true)
+ addSqlFunctionMethod(IS_JSON_OBJECT, Seq(VARCHAR),
+ BuiltInMethod.IS_JSON_OBJECT.method, argsNullable = true)
+
+ addSqlFunctionMethod(IS_JSON_ARRAY, Seq(CHAR),
+ BuiltInMethod.IS_JSON_ARRAY.method, argsNullable = true)
+ addSqlFunctionMethod(IS_JSON_ARRAY, Seq(VARCHAR),
+ BuiltInMethod.IS_JSON_ARRAY.method, argsNullable = true)
+
+ addSqlFunctionMethod(IS_JSON_SCALAR, Seq(CHAR),
+ BuiltInMethod.IS_JSON_SCALAR.method, argsNullable = true)
+ addSqlFunctionMethod(IS_JSON_SCALAR, Seq(VARCHAR),
+ BuiltInMethod.IS_JSON_SCALAR.method, argsNullable = true)
addSqlFunction(IS_NOT_JSON_VALUE, Seq(CHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_VALUE.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_VALUE.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_VALUE, Seq(VARCHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_VALUE.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_VALUE.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_OBJECT, Seq(CHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_OBJECT.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_OBJECT.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_OBJECT, Seq(VARCHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_OBJECT.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_OBJECT.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_ARRAY, Seq(CHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_ARRAY.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_ARRAY.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_ARRAY, Seq(VARCHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_ARRAY.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_ARRAY.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_SCALAR, Seq(CHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_SCALAR.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_SCALAR.method, argsNullable = true)))
addSqlFunction(IS_NOT_JSON_SCALAR, Seq(VARCHAR),
- new NotCallGen(
- new MethodCallGen(
- BuiltInMethod.IS_JSON_SCALAR.method)))
+ new NotCallGen(new MethodCallGen(BuiltInMethod.IS_JSON_SCALAR.method, argsNullable = true)))
// ----------------------------------------------------------------------------------------------
@@ -892,8 +884,9 @@ class FunctionGenerator private(config: TableConfig) {
private def addSqlFunctionMethod(
sqlOperator: SqlOperator,
operandTypes: Seq[LogicalTypeRoot],
- method: Method): Unit = {
- sqlFunctions((sqlOperator, operandTypes)) = new MethodCallGen(method)
+ method: Method,
+ argsNullable: Boolean = false): Unit = {
+ sqlFunctions((sqlOperator, operandTypes)) = new MethodCallGen(method, argsNullable)
}
private def addSqlFunction(
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/MethodCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/MethodCallGen.scala
index 9a8284c..fefbe5a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/MethodCallGen.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/MethodCallGen.scala
@@ -19,51 +19,59 @@
package org.apache.flink.table.planner.codegen.calls
import org.apache.flink.table.planner.codegen.CodeGenUtils.{BINARY_STRING, qualifyMethod}
-import org.apache.flink.table.planner.codegen.GenerateUtils.generateCallIfArgsNotNull
+import org.apache.flink.table.planner.codegen.GenerateUtils.{generateCallIfArgsNotNull, generateCallIfArgsNullable}
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.types.logical.LogicalType
+
import java.lang.reflect.Method
import java.util.TimeZone
-class MethodCallGen(method: Method) extends CallGenerator {
+class MethodCallGen(method: Method, argsNullable: Boolean = false) extends CallGenerator {
override def generate(
ctx: CodeGeneratorContext,
operands: Seq[GeneratedExpression],
returnType: LogicalType): GeneratedExpression = {
- generateCallIfArgsNotNull(ctx, returnType, operands, !method.getReturnType.isPrimitive) {
- originalTerms => {
- val terms = originalTerms.zip(method.getParameterTypes).map { case (term, clazz) =>
- // convert the StringData parameter to String if the method parameter accept String
- if (clazz == classOf[String]) {
- s"$term.toString()"
- } else {
- term
- }
- }
+ if (argsNullable) {
+ generateCallIfArgsNullable(ctx, returnType, operands, !method.getReturnType.isPrimitive) {
+ originalTerms => convertResult(ctx, originalTerms)
+ }
+ } else {
+ generateCallIfArgsNotNull(ctx, returnType, operands, !method.getReturnType.isPrimitive) {
+ originalTerms => convertResult(ctx, originalTerms)
+ }
+ }
+ }
- // generate method invoke code and adapt when it's a time zone related function
- val call = if (terms.length + 1 == method.getParameterCount &&
- method.getParameterTypes()(terms.length) == classOf[TimeZone]) {
- // insert the zoneID parameters for timestamp functions
- val timeZone = ctx.addReusableSessionTimeZone()
- s"""
- |${qualifyMethod(method)}(${terms.mkString(", ")}, $timeZone)
+ private def convertResult(ctx: CodeGeneratorContext, originalTerms: Seq[String]): String = {
+ val terms = originalTerms.zip(method.getParameterTypes).map { case (term, clazz) =>
+ // convert the StringData parameter to String if the method parameter accept String
+ if (clazz == classOf[String]) {
+ s"$term.toString()"
+ } else {
+ term
+ }
+ }
+
+ // generate method invoke code and adapt when it's a time zone related function
+ val call = if (terms.length + 1 == method.getParameterCount &&
+ method.getParameterTypes()(terms.length) == classOf[TimeZone]) {
+ // insert the zoneID parameters for timestamp functions
+ val timeZone = ctx.addReusableSessionTimeZone()
+ s"""
+ |${qualifyMethod(method)}(${terms.mkString(", ")}, $timeZone)
""".stripMargin
- } else {
- s"""
- |${qualifyMethod(method)}(${terms.mkString(", ")})
+ } else {
+ s"""
+ |${qualifyMethod(method)}(${terms.mkString(", ")})
""".stripMargin
- }
-
- // convert String to StringData if the return type is String
- if (method.getReturnType == classOf[String]) {
- s"$BINARY_STRING.fromString($call)"
- } else {
- call
- }
- }
+ }
+ // convert String to StringData if the return type is String
+ if (method.getReturnType == classOf[String]) {
+ s"$BINARY_STRING.fromString($call)"
+ } else {
+ call
}
}
}