This is an automated email from the ASF dual-hosted git repository.
raminqaf 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 15e278160fd [FLINK-40463][table] Widen `JSON_VALUE` and `JSON_QUERY`
`RETURNING` type support (#29506)
15e278160fd is described below
commit 15e278160fd0572ae06f70d073aeb743ee344315
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Wed Oct 21 12:22:57 2026 +0200
[FLINK-40463][table] Widen `JSON_VALUE` and `JSON_QUERY` `RETURNING` type
support (#29506)
[FLINK-40463][table] Widen `JSON_VALUE` and `JSON_QUERY` `RETURNING` type
support
`JSON_VALUE ... RETURNING` now also accepts `TINYINT`, `SMALLINT`,
`BIGINT`, `FLOAT` and `DECIMAL`. `JSON_QUERY ... RETURNING` accepts an `ARRAY`
of any of these scalar types, not just `ARRAY<STRING>`.
JSON strings are coerced to the requested type, for example `"42"` to a
number or `"true"` to `TRUE`. A value that does not fit, such as an overflow or
a non-numeric string, follows the `ON ERROR` behavior.
This closes #29506.
---------
Generated-by: Claude <[email protected]>
Co-authored-by: Chris Johnson <[email protected]>
---
docs/data/sql_functions.yml | 16 +-
docs/data/sql_functions_zh.yml | 22 +-
flink-python/pyflink/table/expression.py | 5 +
.../flink/table/api/internal/BaseExpressions.java | 11 +-
.../functions/sql/SqlJsonQueryFunctionWrapper.java | 9 +-
.../planner/codegen/calls/JsonQueryCallGen.scala | 82 ++-
.../planner/codegen/calls/JsonValueCallGen.scala | 105 ++--
.../planner/functions/CastFunctionITCase.java | 7 +-
.../planner/functions/JsonFunctionsITCase.java | 343 ++++++++++++-
.../src/test/resources/json/json-value.json | 2 +
.../table/data/binary/BinaryStringDataUtil.java | 6 +-
.../table/runtime/functions/SqlJsonUtils.java | 321 +++++++++++-
.../functions/SqlJsonUtilsConversionTest.java | 559 +++++++++++++++++++++
13 files changed, 1413 insertions(+), 75 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 4244ed6a49e..22463dcdf00 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1086,8 +1086,18 @@ json:
* `VARCHAR` / `STRING`
* `BOOLEAN`
+ * `TINYINT`
+ * `SMALLINT`
* `INTEGER`
+ * `BIGINT`
+ * `FLOAT`
* `DOUBLE`
+ * `DECIMAL`
+
+ JSON string values are coerced to the requested numeric or boolean type
(for example `"42"`
+ to a number, or `"true"`/`"t"`/`"yes"`/`"y"`/`"1"` to `TRUE`).
+ Values that do not fit the requested type (overflow, non-numeric string,
non-0/1 number for
+ `BOOLEAN`) are treated as an error and follow the `ON ERROR` behavior.
For empty path expressions or errors a behavior can be defined to either
return `null`, raise
an error or return a defined default value instead. When omitted, the
default is
@@ -1127,7 +1137,11 @@ json:
description: |
Extracts JSON values from a JSON string.
- The result is returned as a `STRING` or `ARRAY<STRING>`. This can be
controlled with the `RETURNING` clause.
+ The result is returned as a `STRING` by default. Using the `RETURNING`
clause it can also be
+ returned as an `ARRAY` whose element type is `STRING` or any of the
scalar types supported by
+ `JSON_VALUE` (`BOOLEAN`, `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT`,
`FLOAT`, `DOUBLE`,
+ `DECIMAL`). Array elements are converted with the same coercion and `ON
ERROR` rules as
+ `JSON_VALUE`.
The `wrappingBehavior` determines whether the extracted value should be
wrapped into an array,
and whether to do so unconditionally or only if the value itself isn't
an array already.
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 74b5b0b9389..d1dc90163fb 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1219,8 +1219,18 @@ json:
* `VARCHAR` / `STRING`
* `BOOLEAN`
+ * `TINYINT`
+ * `SMALLINT`
* `INTEGER`
+ * `BIGINT`
+ * `FLOAT`
* `DOUBLE`
+ * `DECIMAL`
+
+ JSON string values are coerced to the requested numeric or boolean type
(for example `"42"`
+ to a number, or `"true"`/`"t"`/`"yes"`/`"y"`/`"1"` to `TRUE`).
+ Values that do not fit the requested type (overflow, non-numeric string,
non-0/1 number for
+ `BOOLEAN`) are treated as an error and follow the `ON ERROR` behavior.
对于空路径表达式或错误，可以将行为定义为返回 `null`、引发错误或返回定义的默认值。当省略时，默认为 `NULL ON EMPTY` 或
`NULL ON ERROR`。默认值可以是文字或表达式。如果默认值本身引发错误，就会造成 `ON EMPTY` 的错误行为，并引发 `ON
ERROR` 的错误。
@@ -1250,12 +1260,16 @@ json:
-- "right"
JSON_VALUE('{"contains blank": "right"}', 'strict $.[''contains
blank'']' NULL ON EMPTY DEFAULT 'wrong' ON ERROR)
```
- - sql: JSON_QUERY(jsonValue, path [ { WITHOUT | WITH CONDITIONAL | WITH
UNCONDITIONAL } [ ARRAY ] WRAPPER ] [ { NULL | EMPTY ARRAY | EMPTY OBJECT |
ERROR } ON EMPTY ] [ { NULL | EMPTY ARRAY | EMPTY OBJECT | ERROR } ON ERROR ])
- table: STRING.jsonQuery(path [, JsonQueryWrapper [,
JsonQueryOnEmptyOrError, JsonQueryOnEmptyOrError ] ])
+ - sql: JSON_QUERY(jsonValue, path [RETURNING <dataType>] [ { WITHOUT | WITH
CONDITIONAL | WITH UNCONDITIONAL } [ ARRAY ] WRAPPER ] [ { NULL | EMPTY ARRAY |
EMPTY OBJECT | ERROR } ON EMPTY ] [ { NULL | EMPTY ARRAY | EMPTY OBJECT | ERROR
} ON ERROR ])
+ table: STRING.jsonQuery(path [, returnType [, JsonQueryWrapper [,
JsonQueryOnEmptyOrError, JsonQueryOnEmptyOrError ] ] ])
description: |
- 从 JSON 字符串中提取 JSON 值。
+ Extracts JSON values from a JSON string.
- 结果总是以 `STRING` 的形式返回。目前尚不支持 `RETURNING` 子句。
+ The result is returned as a `STRING` by default. Using the `RETURNING`
clause it can also be
+ returned as an `ARRAY` whose element type is `STRING` or any of the
scalar types supported by
+ `JSON_VALUE` (`BOOLEAN`, `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT`,
`FLOAT`, `DOUBLE`,
+ `DECIMAL`). Array elements are converted with the same coercion and `ON
ERROR` rules as
+ `JSON_VALUE`.
`wrappingBehavior` 决定是否将提取的值包装到一个数组中，以及是否无条件地这样做，还是只有当值本身不是数组时才这样做。
diff --git a/flink-python/pyflink/table/expression.py
b/flink-python/pyflink/table/expression.py
index 4adaa96ca4b..0160904e263 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -2305,8 +2305,13 @@ class Expression(Generic[T]):
* `STRING`
* `BOOLEAN`
+ * `TINYINT`
+ * `SMALLINT`
* `INT`
+ * `BIGINT`
+ * `FLOAT`
* `DOUBLE`
+ * `DECIMAL`
For empty path expressions or errors a behavior can be defined to
either return `null`,
raise an error or return a defined default value instead.
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index c4b7f9677d5..29f12604fc8 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -2436,8 +2436,13 @@ public abstract class BaseExpressions<InType, OutType> {
* <ul>
* <li>{@link DataTypes#STRING()}
* <li>{@link DataTypes#BOOLEAN()}
+ * <li>{@link DataTypes#TINYINT()}
+ * <li>{@link DataTypes#SMALLINT()}
* <li>{@link DataTypes#INT()}
+ * <li>{@link DataTypes#BIGINT()}
+ * <li>{@link DataTypes#FLOAT()}
* <li>{@link DataTypes#DOUBLE()}
+ * <li>{@link DataTypes#DECIMAL(int, int)}
* </ul>
*
* <p>For empty path expressions or errors a behavior can be defined to
either return {@code
@@ -2571,7 +2576,11 @@ public abstract class BaseExpressions<InType, OutType> {
* Extracts JSON values from a JSON string.
*
* <p>This follows the ISO/IEC TR 19075-6 specification for JSON support
in SQL. The result is
- * always returned as a {@link DataTypes#STRING()}.
+ * returned as a {@link DataTypes#STRING()} by default. Using {@param
returnType} it can also be
+ * returned as an {@link DataTypes#ARRAY(DataType)} whose element type is
{@code STRING} or any
+ * of the scalar types supported by {@link #jsonValue(String, DataType)}
({@code BOOLEAN},
+ * {@code TINYINT}, {@code SMALLINT}, {@code INTEGER}, {@code BIGINT},
{@code FLOAT}, {@code
+ * DOUBLE}, {@code DECIMAL}).
*
* <p>The {@param wrappingBehavior} determines whether the extracted value
should be wrapped
* into an array, and whether to do so unconditionally or only if the
value itself isn't an
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonQueryFunctionWrapper.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonQueryFunctionWrapper.java
index ddae97fa232..31cacfa55f7 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonQueryFunctionWrapper.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonQueryFunctionWrapper.java
@@ -19,6 +19,8 @@
package org.apache.flink.table.planner.functions.sql;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.runtime.functions.SqlJsonUtils;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.sql.SqlCallBinding;
@@ -89,12 +91,15 @@ public class SqlJsonQueryFunctionWrapper extends
SqlJsonQueryFunction {
private static boolean checkOperandsForArrayReturnType(
boolean throwOnFailure, RelDataType type, SqlCallBinding
callBinding) {
- if (!SqlTypeUtil.isCharacter(type.getComponentType())) {
+ RelDataType elementType = type.getComponentType();
+ if (!SqlTypeUtil.isCharacter(elementType)
+ && !SqlJsonUtils.isSupportedJsonReturningType(
+
FlinkTypeFactory.toLogicalType(elementType).getTypeRoot())) {
if (throwOnFailure) {
throw new ValidationException(
String.format(
"Unsupported array element type '%s' for
RETURNING ARRAY in JSON_QUERY().",
- type.getComponentType()));
+ elementType));
} else {
return false;
}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonQueryCallGen.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonQueryCallGen.scala
index 5eb269ff4f8..ef5e1e2349b 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonQueryCallGen.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonQueryCallGen.scala
@@ -17,14 +17,14 @@
*/
package org.apache.flink.table.planner.codegen.calls
-import org.apache.flink.table.api.{JsonQueryOnEmptyOrError, JsonQueryWrapper,
JsonValueOnEmptyOrError}
+import org.apache.flink.table.api.{JsonQueryOnEmptyOrError, JsonQueryWrapper}
+import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext,
CodeGenException, CodeGenUtils, GeneratedExpression}
import org.apache.flink.table.planner.codegen.CodeGenUtils.{qualifyEnum,
qualifyMethod, BINARY_STRING, GENERIC_ARRAY}
import
org.apache.flink.table.planner.codegen.GenerateUtils.generateCallWithStmtIfArgsNotNull
+import org.apache.flink.table.runtime.functions.SqlJsonUtils
import
org.apache.flink.table.runtime.functions.SqlJsonUtils.JsonQueryReturnType
-import org.apache.flink.table.types.logical.{ArrayType, LogicalType,
LogicalTypeRoot}
-
-import org.apache.calcite.sql.SqlJsonEmptyOrError
+import org.apache.flink.table.types.logical.{ArrayType, DecimalType,
LogicalType, LogicalTypeRoot}
/**
* [[CallGenerator]] for [[BuiltInMethods.JSON_QUERY]].
@@ -72,11 +72,24 @@ class JsonQueryCallGen extends CallGenerator {
val emptyBehavior =
operands(3).literalValue.get.asInstanceOf[JsonQueryOnEmptyOrError]
val errorBehavior =
operands(4).literalValue.get.asInstanceOf[JsonQueryOnEmptyOrError]
val wrapperBehavior =
operands(2).literalValue.get.asInstanceOf[JsonQueryWrapper]
- val jsonQueryReturnType = if (returnType.getTypeRoot ==
LogicalTypeRoot.ARRAY) {
- JsonQueryReturnType.ARRAY
- } else {
- JsonQueryReturnType.STRING
+
+ // Three return paths:
+ // VARCHAR -> STRING -> serialize to StringData
+ // ARRAY<VARCHAR> -> ARRAY -> GenericArrayData of StringData
(existing)
+ // ARRAY<typed T> -> RAW_ARRAY -> Object[] post-converted via
convertJsonArray
+ val (isTypedArray, elementType) = returnType.getTypeRoot match {
+ case LogicalTypeRoot.ARRAY =>
+ val et = returnType.asInstanceOf[ArrayType].getElementType
+ (et.getTypeRoot != LogicalTypeRoot.VARCHAR, et)
+ case _ => (false, null)
+ }
+
+ val jsonQueryReturnType = returnType.getTypeRoot match {
+ case LogicalTypeRoot.ARRAY if isTypedArray =>
JsonQueryReturnType.RAW_ARRAY
+ case LogicalTypeRoot.ARRAY => JsonQueryReturnType.ARRAY
+ case _ => JsonQueryReturnType.STRING
}
+
val terms = Seq(
parsed.resultTerm,
s"${argTerms(1)}.toString()",
@@ -87,29 +100,54 @@ class JsonQueryCallGen extends CallGenerator {
)
val rawResultTerm = CodeGenUtils.newName(ctx, "rawResult")
- val call = s"""
- |Object $rawResultTerm =
- |
${qualifyMethod(BuiltInMethods.JSON_QUERY_PARSED)}(${terms
- .mkString(", ")});
+ val baseCall = s"""
+ |Object $rawResultTerm =
+ |
${qualifyMethod(BuiltInMethods.JSON_QUERY_PARSED)}(${terms
+ .mkString(", ")});
""".stripMargin
- val convertedResult = returnType.getTypeRoot match {
+ val (additionalCall, convertedResult) = returnType.getTypeRoot match
{
case LogicalTypeRoot.VARCHAR =>
-
s"$BINARY_STRING.fromString(java.lang.String.valueOf($rawResultTerm))"
+ ("",
s"$BINARY_STRING.fromString(java.lang.String.valueOf($rawResultTerm))")
+ case LogicalTypeRoot.ARRAY if !isTypedArray =>
+ ("", s"($GENERIC_ARRAY) $rawResultTerm")
+ case LogicalTypeRoot.ARRAY
+ if
!SqlJsonUtils.isSupportedJsonReturningType(elementType.getTypeRoot) =>
+ throw new CodeGenException(
+ s"Unsupported element type '${elementType.getTypeRoot}' "
+ + "for RETURNING ARRAY in JSON_QUERY().")
case LogicalTypeRoot.ARRAY =>
- val elementType =
returnType.asInstanceOf[ArrayType].getElementType
- if (elementType.getTypeRoot == LogicalTypeRoot.VARCHAR) {
- s"($GENERIC_ARRAY) $rawResultTerm"
- } else {
- throw new CodeGenException(
- s"Unsupported array element type '$elementType' for
RETURNING ARRAY in JSON_QUERY().")
+ val jsonUtils =
+ classOf[SqlJsonUtils].getName
+ val typeRootEnum =
+
s"${classOf[LogicalTypeRoot].getName}.${elementType.getTypeRoot.name()}"
+ val (precisionStr, scaleStr) = elementType.getTypeRoot match {
+ case LogicalTypeRoot.DECIMAL =>
+ val dt = elementType.asInstanceOf[DecimalType]
+ (dt.getPrecision.toString, dt.getScale.toString)
+ case _ => ("0", "0")
}
+ val errorBehaviorEnum = qualifyEnum(errorBehavior)
+
+ val elementNullableStr = elementType.isNullable.toString
+ val convertedTerm = CodeGenUtils.newName(ctx, "convertedArr")
+ val conversionCode =
+ s"""
+ |$GENERIC_ARRAY $convertedTerm =
$jsonUtils.convertJsonArray(
+ | $rawResultTerm, $typeRootEnum,
+ | $precisionStr, $scaleStr,
+ | $elementNullableStr,
+ | $errorBehaviorEnum);
+ """.stripMargin
+
+ (conversionCode, s"$convertedTerm")
case _ =>
- throw new CodeGenException(
+ throw new ValidationException(
s"Unsupported type '$returnType' "
- + "for RETURNING in JSON_VALUE().")
+ + "for RETURNING in JSON_QUERY().")
}
+ val call = baseCall + additionalCall
val result = s"($rawResultTerm == null) ? null : ($convertedResult)"
(call, result)
}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonValueCallGen.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonValueCallGen.scala
index 321a6d072c3..1bf570f3312 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonValueCallGen.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonValueCallGen.scala
@@ -19,9 +19,10 @@ package org.apache.flink.table.planner.codegen.calls
import org.apache.flink.table.api.JsonValueOnEmptyOrError
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext,
CodeGenException, CodeGenUtils, GeneratedExpression}
-import org.apache.flink.table.planner.codegen.CodeGenUtils.{qualifyEnum,
qualifyMethod, BINARY_STRING}
+import
org.apache.flink.table.planner.codegen.CodeGenUtils.{isPrimitiveNullable,
primitiveDefaultValue, primitiveTypeTermForType, qualifyEnum, qualifyMethod,
BINARY_STRING}
import
org.apache.flink.table.planner.codegen.GenerateUtils.generateCallWithStmtIfArgsNotNull
-import org.apache.flink.table.types.logical.{LogicalType, LogicalTypeRoot}
+import org.apache.flink.table.runtime.functions.SqlJsonUtils
+import org.apache.flink.table.types.logical.{DecimalType, LogicalType,
LogicalTypeRoot}
import org.apache.calcite.sql.SqlJsonEmptyOrError
@@ -66,43 +67,81 @@ class JsonValueCallGen extends CallGenerator {
val parsed = JsonParseReuse.parseSharedInput(ctx, operands)
- generateCallWithStmtIfArgsNotNull(ctx, returnType, operands,
resultNullable = true) {
- argTerms =>
- {
- val emptyBehavior = getBehavior(operands, SqlJsonEmptyOrError.EMPTY)
- val errorBehavior = getBehavior(operands, SqlJsonEmptyOrError.ERROR)
+ val generated =
+ generateCallWithStmtIfArgsNotNull(ctx, returnType, operands,
resultNullable = true) {
+ argTerms =>
+ {
+ val emptyBehavior = getBehavior(operands,
SqlJsonEmptyOrError.EMPTY)
+ val errorBehavior = getBehavior(operands,
SqlJsonEmptyOrError.ERROR)
- val terms = Seq(
- parsed.resultTerm,
- s"${argTerms(1)}.toString()",
- qualifyEnum(emptyBehavior._1),
- emptyBehavior._2,
- qualifyEnum(errorBehavior._1),
- errorBehavior._2
- )
+ val terms = Seq(
+ parsed.resultTerm,
+ s"${argTerms(1)}.toString()",
+ qualifyEnum(emptyBehavior._1),
+ emptyBehavior._2,
+ qualifyEnum(errorBehavior._1),
+ errorBehavior._2
+ )
- val rawResultTerm = CodeGenUtils.newName(ctx, "rawResult")
- val call = s"""
- |Object $rawResultTerm =
- |
${qualifyMethod(BuiltInMethods.JSON_VALUE)}(${terms
- .mkString(", ")});
+ val rawResultTerm = CodeGenUtils.newName(ctx, "rawResult")
+ val baseCall = s"""
+ |Object $rawResultTerm =
+ |
${qualifyMethod(BuiltInMethods.JSON_VALUE)}(${terms
+ .mkString(", ")});
""".stripMargin
- val convertedResult = returnType.getTypeRoot match {
- case LogicalTypeRoot.VARCHAR =>
-
s"$BINARY_STRING.fromString(java.lang.String.valueOf($rawResultTerm))"
- case LogicalTypeRoot.BOOLEAN => s"(java.lang.Boolean)
$rawResultTerm"
- case LogicalTypeRoot.INTEGER => s"(java.lang.Integer)
$rawResultTerm"
- case LogicalTypeRoot.DOUBLE => s"((java.math.BigDecimal)
$rawResultTerm).doubleValue()"
- case _ =>
- throw new CodeGenException(
- s"Unsupported type '$returnType' "
- + "for RETURNING in JSON_VALUE().")
+ returnType.getTypeRoot match {
+ case LogicalTypeRoot.VARCHAR =>
+ val result =
+ s"($rawResultTerm == null) ? null : " +
+
s"($BINARY_STRING.fromString(java.lang.String.valueOf($rawResultTerm)))"
+ (baseCall, result)
+ case typeRoot if
!SqlJsonUtils.isSupportedJsonReturningType(typeRoot) =>
+ throw new CodeGenException(
+ s"Unsupported type '$returnType' for RETURNING in
JSON_VALUE().")
+ case _ =>
+ val jsonUtils =
+ classOf[SqlJsonUtils].getName
+ val typeRootEnum =
+
s"${classOf[LogicalTypeRoot].getName}.${returnType.getTypeRoot.name()}"
+ val (precisionStr, scaleStr) = returnType.getTypeRoot match {
+ case LogicalTypeRoot.DECIMAL =>
+ val dt = returnType.asInstanceOf[DecimalType]
+ (dt.getPrecision.toString, dt.getScale.toString)
+ case _ => ("0", "0")
+ }
+ val errorBehaviorEnum =
+
qualifyEnum(errorBehavior._1.asInstanceOf[JsonValueOnEmptyOrError])
+ val defaultExpr =
+ if (errorBehavior._2 != null) errorBehavior._2 else "null"
+
+ val boxedType = CodeGenUtils.boxedTypeTermForType(returnType)
+ val convertedTerm = CodeGenUtils.newName(ctx, "converted")
+ val call = baseCall +
+ s"""
+ |$boxedType $convertedTerm = ($boxedType)
$jsonUtils.convertJsonScalar(
+ | $rawResultTerm, $typeRootEnum,
+ | $precisionStr, $scaleStr,
+ | $errorBehaviorEnum, $defaultExpr);
+ """.stripMargin
+ (call, s"$convertedTerm")
+ }
}
+ }
- val result = s"($rawResultTerm == null) ? null : ($convertedResult)"
- (call, result)
- }
+ if (isPrimitiveNullable(returnType)) {
+ generated
+ } else {
+ val unboxedTerm =
+ ctx.addReusableLocalVariable(primitiveTypeTermForType(returnType),
"jsonValueUnboxed")
+ val unboxedCode =
+ s"""
+ |${generated.code}
+ |$unboxedTerm = ${generated.nullTerm}
+ | ? ${primitiveDefaultValue(returnType)}
+ | : ${generated.resultTerm};
+ |""".stripMargin
+ generated.copy(resultTerm = unboxedTerm, code = unboxedCode)
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
index e3eba6e9a73..6163ee022f3 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.planner.functions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
-import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
@@ -1252,12 +1251,12 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
return Arrays.asList(
CastTestSpecBuilder.testCastTo(BOOLEAN())
.fromCase(BOOLEAN(), null, null)
- .failRuntime(CHAR(3), "foo", TableException.class)
+ .failRuntime(CHAR(3), "foo",
TableRuntimeException.class)
.fromCase(CHAR(4), "true", true)
.fromCase(VARCHAR(5), "FalsE", false)
- .failRuntime(STRING(), "Apache Flink",
TableException.class)
+ .failRuntime(STRING(), "Apache Flink",
TableRuntimeException.class)
.fromCase(STRING(), "TRUE", true)
- .failRuntime(STRING(), "", TableException.class)
+ .failRuntime(STRING(), "", TableRuntimeException.class)
.fromCase(BOOLEAN(), true, true)
.fromCase(BOOLEAN(), false, false)
// Not supported - no fix
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
index 2cd54053210..aaeb24f6a0a 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
@@ -38,6 +38,7 @@ import org.apache.commons.io.IOUtils;
import java.io.IOException;
import java.io.InputStream;
+import java.math.BigDecimal;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
@@ -51,18 +52,22 @@ import java.util.Map;
import java.util.stream.Stream;
import static org.apache.flink.table.api.DataTypes.ARRAY;
+import static org.apache.flink.table.api.DataTypes.BIGINT;
import static org.apache.flink.table.api.DataTypes.BINARY;
import static org.apache.flink.table.api.DataTypes.BOOLEAN;
import static org.apache.flink.table.api.DataTypes.BYTES;
import static org.apache.flink.table.api.DataTypes.DECIMAL;
import static org.apache.flink.table.api.DataTypes.DOUBLE;
import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.FLOAT;
import static org.apache.flink.table.api.DataTypes.INT;
import static org.apache.flink.table.api.DataTypes.MAP;
import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.SMALLINT;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static
org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
+import static org.apache.flink.table.api.DataTypes.TINYINT;
import static org.apache.flink.table.api.DataTypes.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.apache.flink.table.api.Expressions.$;
@@ -482,6 +487,11 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
"JSON_VALUE(f0, '$.activated' RETURNING BOOLEAN)",
true,
BOOLEAN())
+ // string-to-boolean coercion matches CAST
(BinaryStringDataUtil.toBoolean)
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": \"y\"}', '$.a' RETURNING
BOOLEAN)", true, BOOLEAN())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": \"N\"}', '$.a' RETURNING
BOOLEAN)", false, BOOLEAN())
.testResult(
$("f0").jsonValue("$.age", INT()),
"JSON_VALUE(f0, '$.age' RETURNING INT)",
@@ -497,7 +507,36 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
"JSON_VALUE(f0, '$.longBalance' RETURNING DOUBLE)",
123456789.987654321,
DOUBLE())
-
+ .testResult(
+ $("f0").jsonValue("$.age", TINYINT()),
+ "JSON_VALUE(f0, '$.age' RETURNING TINYINT)",
+ (byte) 42,
+ TINYINT())
+ .testResult(
+ $("f0").jsonValue("$.age", SMALLINT()),
+ "JSON_VALUE(f0, '$.age' RETURNING SMALLINT)",
+ (short) 42,
+ SMALLINT())
+ .testResult(
+ $("f0").jsonValue("$.age", BIGINT()),
+ "JSON_VALUE(f0, '$.age' RETURNING BIGINT)",
+ 42L,
+ BIGINT())
+ .testResult(
+ $("f0").jsonValue("$.bigCount", BIGINT()),
+ "JSON_VALUE(f0, '$.bigCount' RETURNING BIGINT)",
+ 9999999999L,
+ BIGINT())
+ .testResult(
+ $("f0").jsonValue("$.balance", FLOAT()),
+ "JSON_VALUE(f0, '$.balance' RETURNING FLOAT)",
+ 13.37f,
+ FLOAT())
+ .testResult(
+ $("f0").jsonValue("$.balance", DECIMAL(10, 2)),
+ "JSON_VALUE(f0, '$.balance' RETURNING DECIMAL(10, 2))",
+ new BigDecimal("13.37"),
+ DECIMAL(10, 2))
// ON EMPTY / ON ERROR
.testResult(
$("f0").jsonValue(
@@ -522,6 +561,16 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
"JSON_VALUE(f0, 'lax $.invalid' RETURNING INTEGER
DEFAULT 42 ON EMPTY ERROR ON ERROR)",
42,
INT())
+ // A DECIMAL-typed default into a non-DECIMAL RETURNING type
must be converted,
+ // not dropped to NULL
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1}', 'lax $.missing' RETURNING
INTEGER DEFAULT 1.5 ON EMPTY)",
+ 1,
+ INT())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1}', 'lax $.missing' RETURNING
BIGINT DEFAULT 1.5 ON EMPTY)",
+ 1L,
+ BIGINT())
.testResult(
$("f0").jsonValue(
"strict $.invalid",
@@ -545,7 +594,149 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase
{
"JSON_VALUE(f0, 'strict $.invalid' RETURNING INTEGER
NULL ON EMPTY DEFAULT 42 ON ERROR)",
42,
INT())
+ // JSON null at a valid path triggers ON EMPTY (default: NULL)
+ .testResult(
+ $("f0").jsonValue("$.nullField", INT()),
+ "JSON_VALUE(f0, '$.nullField' RETURNING INTEGER)",
+ null,
+ INT())
+
+ // Type mismatch: string value cast to numeric triggers ON
ERROR
+ .testResult(
+ $("f0").jsonValue(
+ "$.type",
+ INT(),
+ JsonValueOnEmptyOrError.NULL,
+ null,
+ JsonValueOnEmptyOrError.DEFAULT,
+ 42),
+ "JSON_VALUE(f0, '$.type' RETURNING INTEGER DEFAULT 42
ON ERROR)",
+ 42,
+ INT())
+ .testSqlRuntimeError(
+ "JSON_VALUE(f0, '$.type' RETURNING INTEGER ERROR ON
ERROR)",
+ TableRuntimeException.class,
+ "Cannot cast")
+
+ // Numeric overflow triggers ON ERROR (not silent wrapping)
+ .testResult(
+ $("f0").jsonValue("$.bigCount", INT()),
+ "JSON_VALUE(f0, '$.bigCount' RETURNING INTEGER)",
+ null,
+ INT())
+ // Fractional truncation toward zero (13.89 -> 13, -13.89 ->
-13):
+ // MySQL truncates, PostgreSQL errors. We match MySQL behavior
+ // (CAST(JSON_UNQUOTE(JSON_EXTRACT(...)) AS type)).
+ .testResult(
+ $("f0").jsonValue("$.balance", INT()),
+ "JSON_VALUE(f0, '$.balance' RETURNING INTEGER)",
+ 13,
+ INT())
+ // DECIMAL precision overflow triggers ON ERROR
+ .testResult(
+ $("f0").jsonValue("$.longBalance", DECIMAL(5, 2)),
+ "JSON_VALUE(f0, '$.longBalance' RETURNING DECIMAL(5,
2))",
+ null,
+ DECIMAL(5, 2))
+ // String-literal DEFAULT with numeric RETURNING type (codegen
compile regression)
+ .testSqlResult(
+ "JSON_VALUE(f0, '$.longBalance' RETURNING DECIMAL(5,
2) DEFAULT '0.00' ON ERROR)",
+ new BigDecimal("0.00"),
+ DECIMAL(5, 2))
+ // Non-numeric JSON value RETURNING DECIMAL must route to ON
ERROR, not throw
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": true}', '$.a' RETURNING
DECIMAL(5, 2))",
+ null,
+ DECIMAL(5, 2))
+ .testSqlRuntimeError(
+ "JSON_VALUE('{\"a\": true}', '$.a' RETURNING
DECIMAL(5, 2) ERROR ON ERROR)",
+ TableRuntimeException.class,
+ "Cannot cast java.lang.Boolean to DECIMAL")
+ // Floating JSON value out of integer range overflows to ON
ERROR
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1e20}', '$.a' RETURNING
TINYINT)", null, TINYINT())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1e20}', '$.a' RETURNING
SMALLINT)", null, SMALLINT())
+ .testSqlResult("JSON_VALUE('{\"a\": 1e20}', '$.a' RETURNING
INTEGER)", null, INT())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1e20}', '$.a' RETURNING BIGINT)",
null, BIGINT())
+ .testSqlRuntimeError(
+ "JSON_VALUE('{\"a\": 1e20}', '$.a' RETURNING BIGINT
ERROR ON ERROR)",
+ TableRuntimeException.class,
+ "Cannot cast java.math.BigDecimal to BIGINT")
+
+ // Outer CAST on JSON_VALUE result (boxed-to-primitive codegen)
+ .testSqlResult(
+ "CAST(JSON_VALUE(f0, '$.balance' RETURNING DOUBLE) AS
BIGINT)",
+ 13L,
+ BIGINT())
+ .testSqlResult(
+ "CAST(JSON_VALUE(f0, '$.age' RETURNING INTEGER) AS
DOUBLE)", 42.0, DOUBLE())
+ .testSqlResult(
+ "CAST(JSON_VALUE(f0, '$.age' RETURNING INTEGER) AS
BIGINT)", 42L, BIGINT())
+ .testSqlResult(
+ "CAST(JSON_VALUE(f0, '$.balance' RETURNING DOUBLE) AS
INTEGER)", 13, INT())
+ .testSqlResult(
+ "CAST(JSON_VALUE(f0, '$.longBalance' RETURNING DOUBLE)
AS BIGINT)",
+ 123456789L,
+ BIGINT())
+
+ // RETURNING BOOLEAN used directly as a condition: a null
condition takes ELSE
+ .testSqlResult(
+ "IF(JSON_VALUE(f0, '$.activated' RETURNING BOOLEAN),
1, 2)", 1, INT())
+ .testSqlResult("IF(JSON_VALUE(f0, '$.missing' RETURNING
BOOLEAN), 1, 2)", 2, INT())
+
+ // A real value equal to the primitive default is kept, a null
stays null
+ .testSqlResult("JSON_VALUE('{\"a\": -1}', '$.a' RETURNING
INTEGER)", -1, INT())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": false}', '$.a' RETURNING
BOOLEAN)", false, BOOLEAN())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": false}', '$.missing' RETURNING
BOOLEAN)",
+ null,
+ BOOLEAN())
+
+ // Non-boolean value RETURNING BOOLEAN: default NULL ON ERROR
yields null,
+ // explicit ERROR ON ERROR fails
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 1.9}', '$.a' RETURNING BOOLEAN)",
null, BOOLEAN())
+ .testSqlRuntimeError(
+ "JSON_VALUE('{\"a\": 1.9}', '$.a' RETURNING BOOLEAN
ERROR ON ERROR)",
+ TableRuntimeException.class,
+ "Cannot cast java.math.BigDecimal to BOOLEAN")
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": \"maybe\"}', '$.a' RETURNING
BOOLEAN)",
+ null,
+ BOOLEAN())
+ .testSqlRuntimeError(
+ "JSON_VALUE('{\"a\": \"maybe\"}', '$.a' RETURNING
BOOLEAN ERROR ON ERROR)",
+ TableRuntimeException.class,
+ "Cannot cast java.lang.String to BOOLEAN")
+ // Equality on typed RETURNING results from different paths
+ // (distinct objects, same value -- must use value comparison,
not reference)
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 99999, \"b\": 99999}', '$.a'
RETURNING INT) = JSON_VALUE('{\"a\": 99999, \"b\": 99999}', '$.b' RETURNING
INT)",
+ true,
+ BOOLEAN())
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 13.37, \"b\": 13.37}', '$.a'
RETURNING DOUBLE) = JSON_VALUE('{\"a\": 13.37, \"b\": 13.37}', '$.b' RETURNING
DOUBLE)",
+ true,
+ BOOLEAN())
+ // 200 is outside Short cache (-128..127), so boxed == would
fail
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 200, \"b\": 200}', '$.a'
RETURNING SMALLINT) = JSON_VALUE('{\"a\": 200, \"b\": 200}', '$.b' RETURNING
SMALLINT)",
+ true,
+ BOOLEAN())
+ // 9999999999 is outside Long cache (-128..127)
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 9999999999, \"b\": 9999999999}',
'$.a' RETURNING BIGINT) = JSON_VALUE('{\"a\": 9999999999, \"b\": 9999999999}',
'$.b' RETURNING BIGINT)",
+ true,
+ BOOLEAN())
+ // Float has no JVM cache; exercises boxed Float equality
codegen
+ .testSqlResult(
+ "JSON_VALUE('{\"a\": 13.37, \"b\": 13.37}', '$.a'
RETURNING FLOAT) = JSON_VALUE('{\"a\": 13.37, \"b\": 13.37}', '$.b' RETURNING
FLOAT)",
+ true,
+ BOOLEAN())
// path contains blank characters.
.testResult(
$("f0").jsonValue(
@@ -726,9 +917,14 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
"JSON_QUERY(f0, '$.b' RETURNING ARRAY<STRING>
WITH CONDITIONAL WRAPPER)",
new String[] {"1", "2"},
DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f0").jsonQuery("$.b", ARRAY(INT()),
CONDITIONAL_ARRAY),
+ "JSON_QUERY(f0, '$.b' RETURNING ARRAY<INTEGER>
WITH CONDITIONAL WRAPPER)",
+ new Integer[] {1, 2},
+ ARRAY(INT()))
.testSqlValidationError(
- "JSON_QUERY(f0, '$.b' RETURNING ARRAY<INTEGER>
WITH CONDITIONAL WRAPPER ERROR ON ERROR)",
- " Unsupported array element type 'INTEGER' for
RETURNING ARRAY in JSON_QUERY()")
+ "JSON_QUERY(f0, '$.n1' RETURNING INTEGER)",
+ "for RETURNING in JSON_QUERY().")
.testResult(
$("f0").jsonQuery("$.a"),
"JSON_QUERY(f0, '$.a')",
@@ -903,7 +1099,146 @@ class JsonFunctionsITCase extends
BuiltInFunctionTestBase {
.testTableApiRuntimeError(
$("f0").jsonQuery("strict $.err10",
WITHOUT_ARRAY, NULL, ERROR),
TableRuntimeException.class,
- "No results for path"));
+ "No results for path"),
+
+ // Typed RETURNING ARRAY<T> support
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_QUERY)
+ .onFieldsWithData(
+ "{\"ints\": [1, 2, 3], \"doubles\": [1.5,
2.5], \"bools\": [true, false], \"withNull\": [1, null, 3], \"bigints\": [1,
9999999999]}")
+ .andDataTypes(STRING())
+ .testResult(
+ $("f0").jsonQuery("$.ints", ARRAY(INT())),
+ "JSON_QUERY(f0, '$.ints' RETURNING
ARRAY<INT>)",
+ new Integer[] {1, 2, 3},
+ ARRAY(INT()))
+ .testResult(
+ $("f0").jsonQuery("$.bigints",
ARRAY(BIGINT())),
+ "JSON_QUERY(f0, '$.bigints' RETURNING
ARRAY<BIGINT>)",
+ new Long[] {1L, 9999999999L},
+ ARRAY(BIGINT()))
+ .testResult(
+ $("f0").jsonQuery("$.doubles",
ARRAY(DOUBLE())),
+ "JSON_QUERY(f0, '$.doubles' RETURNING
ARRAY<DOUBLE>)",
+ new Double[] {1.5, 2.5},
+ ARRAY(DOUBLE()))
+ .testResult(
+ $("f0").jsonQuery("$.bools", ARRAY(BOOLEAN())),
+ "JSON_QUERY(f0, '$.bools' RETURNING
ARRAY<BOOLEAN>)",
+ new Boolean[] {true, false},
+ ARRAY(BOOLEAN()))
+ // nested element is not a BOOLEAN -> ON ERROR (NULL)
+ .testSqlResult(
+ "JSON_QUERY('{\"a\": [[1], true]}', '$.a'
RETURNING ARRAY<BOOLEAN>)",
+ null,
+ ARRAY(BOOLEAN()))
+ .testResult(
+ $("f0").jsonQuery("$.ints",
ARRAY(INT().notNull())),
+ "JSON_QUERY(f0, '$.ints' RETURNING ARRAY<INT
NOT NULL>)",
+ new Integer[] {1, 2, 3},
+ ARRAY(INT().notNull()))
+ .testSqlValidationError(
+ "JSON_QUERY(f0, '$.ints' RETURNING
ARRAY<TIMESTAMP>)",
+ "Unsupported array element type"),
+
+ // Type mismatch with ON ERROR behavior
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_QUERY)
+ .onFieldsWithData("{\"strings\": [\"a\", \"b\"]}",
"{\"empty\": []}")
+ .andDataTypes(STRING(), STRING())
+ .testResult(
+ $("f0").jsonQuery(
+ "$.strings",
+ ARRAY(INT()),
+ WITHOUT_ARRAY,
+ NULL,
+ NULL),
+ "JSON_QUERY(f0, '$.strings' RETURNING
ARRAY<INT> NULL ON ERROR)",
+ null,
+ ARRAY(INT()))
+ .testResult(
+ $("f0").jsonQuery(
+ "$.strings",
+ ARRAY(INT()),
+ WITHOUT_ARRAY,
+ NULL,
+ EMPTY_ARRAY),
+ "JSON_QUERY(f0, '$.strings' RETURNING
ARRAY<INT> EMPTY ARRAY ON ERROR)",
+ new Integer[] {},
+ ARRAY(INT()))
+ .testSqlRuntimeError(
+ "JSON_QUERY(f0, '$.strings' RETURNING
ARRAY<INT> ERROR ON ERROR)",
+ TableRuntimeException.class,
+ "Cannot parse string 'a' as INTEGER")
+
+ // Empty JSON array and missing path
+ .testResult(
+ $("f1").jsonQuery("$.empty", ARRAY(INT())),
+ "JSON_QUERY(f1, '$.empty' RETURNING
ARRAY<INT>)",
+ new Integer[] {},
+ ARRAY(INT()))
+ .testResult(
+ $("f0").jsonQuery(
+ "lax $.missing",
+ ARRAY(INT()),
+ WITHOUT_ARRAY,
+ NULL,
+ NULL),
+ "JSON_QUERY(f0, 'lax $.missing' RETURNING
ARRAY<INT> NULL ON EMPTY)",
+ null,
+ ARRAY(INT()))
+ .testResult(
+ $("f0").jsonQuery(
+ "lax $.missing",
+ ARRAY(INT()),
+ WITHOUT_ARRAY,
+ EMPTY_ARRAY,
+ NULL),
+ "JSON_QUERY(f0, 'lax $.missing' RETURNING
ARRAY<INT> EMPTY ARRAY ON EMPTY)",
+ new Integer[] {},
+ ARRAY(INT())),
+
+ // ARRAY<BIGINT> near Long.MAX_VALUE boundary
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_QUERY)
+ .onFieldsWithData("{\"big\": [9223372036854775807,
-9223372036854775808]}")
+ .andDataTypes(STRING())
+ .testResult(
+ $("f0").jsonQuery("$.big", ARRAY(BIGINT())),
+ "JSON_QUERY(f0, '$.big' RETURNING
ARRAY<BIGINT>)",
+ new Long[] {Long.MAX_VALUE, Long.MIN_VALUE},
+ ARRAY(BIGINT())),
+
+ // Edge cases: quoted number strings, nested arrays
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_VALUE)
+ .onFieldsWithData("{\"v\":\"42\"}", "{\"neg\":-100}")
+ .andDataTypes(STRING(), STRING())
+ // Jayway returns String "42" -- string is parsed as
number (MySQL/PG
+ // compatible)
+ .testResult(
+ $("f0").jsonValue(
+ "$.v",
+ INT(),
+ JsonValueOnEmptyOrError.NULL,
+ null,
+ JsonValueOnEmptyOrError.NULL,
+ null),
+ "JSON_VALUE(f0, '$.v' RETURNING INTEGER NULL
ON ERROR)",
+ 42,
+ INT())
+ // Negative value within TINYINT range
+ .testResult(
+ $("f1").jsonValue("$.neg", TINYINT()),
+ "JSON_VALUE(f1, '$.neg' RETURNING TINYINT)",
+ (byte) -100,
+ TINYINT()),
+
+ // Nested arrays: inner arrays are Lists, not Numbers
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_QUERY)
+ .onFieldsWithData("{\"a\":[[1,2],[3,4]]}")
+ .andDataTypes(STRING())
+ .testResult(
+ $("f0").jsonQuery("$.a", ARRAY(INT()),
WITHOUT_ARRAY, NULL, NULL),
+ "JSON_QUERY(f0, '$.a' RETURNING ARRAY<INT>
NULL ON ERROR)",
+ null,
+ ARRAY(INT())));
}
private static List<TestSetSpec> jsonStringSpec() {
diff --git
a/flink-table/flink-table-planner/src/test/resources/json/json-value.json
b/flink-table/flink-table-planner/src/test/resources/json/json-value.json
index ca934d761c9..7a6e6f117b8 100644
--- a/flink-table/flink-table-planner/src/test/resources/json/json-value.json
+++ b/flink-table/flink-table-planner/src/test/resources/json/json-value.json
@@ -8,5 +8,7 @@
"roles": ["user", "viewer"],
"balance": 13.37,
"longBalance": 123456789.987654321,
+ "bigCount": 9999999999,
+ "nullField": null,
"contains blank": "right"
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java
index 490dd3324aa..e73160abf73 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.DecimalDataUtils;
import org.apache.flink.table.data.StringData;
@@ -155,7 +155,7 @@ public class BinaryStringDataUtil {
}
/** Parse a {@link StringData} to boolean. */
- public static boolean toBoolean(BinaryStringData str) throws
TableException {
+ public static boolean toBoolean(BinaryStringData str) {
BinaryStringData lowerCase = str.toLowerCase();
if (TRUE_STRINGS.contains(lowerCase)) {
return true;
@@ -163,7 +163,7 @@ public class BinaryStringDataUtil {
if (FALSE_STRINGS.contains(lowerCase)) {
return false;
}
- throw new TableException("Cannot parse '" + str + "' as BOOLEAN.");
+ throw new TableRuntimeException("Cannot cast '" + str + "' as
BOOLEAN.");
}
/** Calculate the hash value of the given bytes use {@link MessageDigest}.
*/
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
index b524854b574..b319d519e73 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
@@ -24,9 +24,14 @@ import org.apache.flink.table.api.JsonQueryOnEmptyOrError;
import org.apache.flink.table.api.JsonQueryWrapper;
import org.apache.flink.table.api.JsonValueOnEmptyOrError;
import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.DecimalDataUtils;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
import org.apache.flink.table.data.binary.StringUtf8Utils;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
import org.apache.flink.types.variant.Variant;
@@ -51,9 +56,14 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.Arra
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeFactory;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+import javax.annotation.Nullable;
+
import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.math.BigInteger;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -244,7 +254,8 @@ public class SqlJsonUtils {
public enum JsonQueryReturnType {
STRING,
- ARRAY
+ ARRAY,
+ RAW_ARRAY
}
public static Object jsonQuery(
@@ -336,6 +347,9 @@ public class SqlJsonUtils {
}
return new GenericArrayData(arr);
+ case RAW_ARRAY:
+ final List<Object> rawList = (List<Object>) value;
+ return rawList.toArray();
default:
throw new TableRuntimeException("illegal return
type");
}
@@ -358,6 +372,8 @@ public class SqlJsonUtils {
switch (returnType) {
case ARRAY:
return new GenericArrayData(new StringData[0]);
+ case RAW_ARRAY:
+ return new Object[0];
case STRING:
return "[]";
default:
@@ -384,6 +400,8 @@ public class SqlJsonUtils {
switch (returnType) {
case ARRAY:
return new GenericArrayData(new StringData[0]);
+ case RAW_ARRAY:
+ return new Object[0];
case STRING:
return "[]";
default:
@@ -838,6 +856,307 @@ public class SqlJsonUtils {
}
}
+ // --- Type conversion for JSON_VALUE / JSON_QUERY RETURNING ---
+
+ private static final EnumSet<LogicalTypeRoot>
SUPPORTED_JSON_RETURNING_TYPES =
+ EnumSet.of(
+ LogicalTypeRoot.BOOLEAN,
+ LogicalTypeRoot.TINYINT,
+ LogicalTypeRoot.SMALLINT,
+ LogicalTypeRoot.INTEGER,
+ LogicalTypeRoot.BIGINT,
+ LogicalTypeRoot.FLOAT,
+ LogicalTypeRoot.DOUBLE,
+ LogicalTypeRoot.DECIMAL);
+
+ public static boolean isSupportedJsonReturningType(LogicalTypeRoot
typeRoot) {
+ return SUPPORTED_JSON_RETURNING_TYPES.contains(typeRoot);
+ }
+
+ public static Object convertJsonScalar(
+ Object raw,
+ LogicalTypeRoot typeRoot,
+ int precision,
+ int scale,
+ JsonValueOnEmptyOrError errorBehavior,
+ Object defaultValue) {
+ if (raw == null) {
+ return null;
+ }
+ try {
+ return convertToType(raw, typeRoot, precision, scale);
+ } catch (TableRuntimeException e) {
+ switch (errorBehavior) {
+ case NULL:
+ return null;
+ case DEFAULT:
+ return convertDefault(defaultValue, typeRoot, precision,
scale);
+ case ERROR:
+ throw new TableRuntimeException(
+ "Cannot cast " + raw.getClass().getName() + " to "
+ typeRoot, e);
+ default:
+ throw new TableRuntimeException(
+ "Unreachable: unknown error behavior " +
errorBehavior);
+ }
+ }
+ }
+
+ public static GenericArrayData convertJsonArray(
+ Object rawResult,
+ LogicalTypeRoot elementTypeRoot,
+ int precision,
+ int scale,
+ boolean elementNullable,
+ JsonQueryOnEmptyOrError errorBehavior) {
+ if (rawResult == null) {
+ return null;
+ }
+ try {
+ Object[] rawArr = (Object[]) rawResult;
+ Object[] converted = new Object[rawArr.length];
+ for (int i = 0; i < rawArr.length; i++) {
+ if (rawArr[i] != null) {
+ converted[i] = convertToType(rawArr[i], elementTypeRoot,
precision, scale);
+ } else if (!elementNullable) {
+ throw new TableRuntimeException(
+ "Null element at index " + i + " in NOT NULL
array");
+ }
+ }
+ return new GenericArrayData(converted);
+ } catch (TableRuntimeException e) {
+ switch (errorBehavior) {
+ case NULL:
+ return null;
+ case EMPTY_ARRAY:
+ return new GenericArrayData(new Object[0]);
+ case ERROR:
+ throw new TableRuntimeException(
+ "JSON_QUERY array element conversion failed", e);
+ default:
+ return null;
+ }
+ }
+ }
+
+ private static Object convertToType(
+ Object raw, LogicalTypeRoot typeRoot, int precision, int scale) {
+ if (raw instanceof DecimalData) {
+ return convertToType(((DecimalData) raw).toBigDecimal(), typeRoot,
precision, scale);
+ }
+ if (raw instanceof StringData) {
+ return convertToType(raw.toString(), typeRoot, precision, scale);
+ }
+ if (raw instanceof String) {
+ if (typeRoot == LogicalTypeRoot.BOOLEAN) {
+ return BinaryStringDataUtil.toBoolean(
+ BinaryStringData.fromString(((String) raw).trim()));
+ }
+ try {
+ String trimmed = ((String) raw).trim();
+ return convertToType(new BigDecimal(trimmed), typeRoot,
precision, scale);
+ } catch (NumberFormatException e) {
+ throw new TableRuntimeException(
+ "Cannot parse string '" + raw + "' as " + typeRoot, e);
+ }
+ }
+ switch (typeRoot) {
+ case BOOLEAN:
+ if (raw instanceof Number) {
+ BigDecimal bd =
+ raw instanceof BigDecimal
+ ? (BigDecimal) raw
+ : new BigDecimal(raw.toString());
+ if (BigDecimal.ZERO.compareTo(bd) == 0) {
+ return false;
+ }
+ if (BigDecimal.ONE.compareTo(bd) == 0) {
+ return true;
+ }
+ throw new TableRuntimeException("Cannot convert " + raw +
" to BOOLEAN");
+ }
+ if (raw instanceof Boolean) {
+ return raw;
+ }
+ throw new TableRuntimeException(
+ "Cannot convert " + raw.getClass().getName() + " to
BOOLEAN");
+ case TINYINT:
+ return toCheckedByte(asNumber(raw, typeRoot));
+ case SMALLINT:
+ return toCheckedShort(asNumber(raw, typeRoot));
+ case INTEGER:
+ return toCheckedInt(asNumber(raw, typeRoot));
+ case BIGINT:
+ return toCheckedLong(asNumber(raw, typeRoot));
+ case FLOAT:
+ return toCheckedFloat(asNumber(raw, typeRoot));
+ case DOUBLE:
+ return toCheckedDouble(asNumber(raw, typeRoot));
+ case DECIMAL:
+ return toCheckedDecimal(raw, precision, scale);
+ default:
+ throw new TableRuntimeException(
+ "Unsupported type for JSON conversion: " + typeRoot);
+ }
+ }
+
+ private static Object convertDefault(
+ Object defaultValue, LogicalTypeRoot typeRoot, int precision, int
scale) {
+ if (defaultValue == null) {
+ return null;
+ }
+ try {
+ return convertToType(defaultValue, typeRoot, precision, scale);
+ } catch (TableRuntimeException e) {
+ throw new TableRuntimeException(
+ "Default value " + defaultValue + " cannot be represented
as " + typeRoot, e);
+ }
+ }
+
+ private static @Nullable BigInteger toBigIntegerTruncated(Number n) {
+ if (n instanceof BigDecimal) {
+ final BigDecimal bd = (BigDecimal) n;
+ final int integerDigits = bd.precision() - bd.scale();
+ if (integerDigits <= 0) {
+ return BigInteger.ZERO;
+ }
+ if (integerDigits > 19) {
+ throw new TableRuntimeException("Value " + n + " is out of
range");
+ }
+ return bd.toBigInteger();
+ }
+ if (n instanceof BigInteger) {
+ return (BigInteger) n;
+ }
+ if (n instanceof Double || n instanceof Float) {
+ double d = n.doubleValue();
+ if (Double.isNaN(d) || Double.isInfinite(d)) {
+ throw new TableRuntimeException("Value " + n + " is out of
range");
+ }
+ return new BigDecimal(d).toBigInteger();
+ }
+ return null;
+ }
+
+ static byte toCheckedByte(Number n) {
+ BigInteger bi = toBigIntegerTruncated(n);
+ if (bi != null) {
+ if (bi.compareTo(BigInteger.valueOf(Byte.MAX_VALUE)) > 0
+ || bi.compareTo(BigInteger.valueOf(Byte.MIN_VALUE)) < 0) {
+ throw new TableRuntimeException("Value " + n + " is out of
range for TINYINT");
+ }
+ return bi.byteValue();
+ }
+ long v = n.longValue();
+ if (v < Byte.MIN_VALUE || v > Byte.MAX_VALUE) {
+ throw new TableRuntimeException("Value " + n + " is out of range
for TINYINT");
+ }
+ return (byte) v;
+ }
+
+ static short toCheckedShort(Number n) {
+ BigInteger bi = toBigIntegerTruncated(n);
+ if (bi != null) {
+ if (bi.compareTo(BigInteger.valueOf(Short.MAX_VALUE)) > 0
+ || bi.compareTo(BigInteger.valueOf(Short.MIN_VALUE)) < 0) {
+ throw new TableRuntimeException("Value " + n + " is out of
range for SMALLINT");
+ }
+ return bi.shortValue();
+ }
+ long v = n.longValue();
+ if (v < Short.MIN_VALUE || v > Short.MAX_VALUE) {
+ throw new TableRuntimeException("Value " + n + " is out of range
for SMALLINT");
+ }
+ return (short) v;
+ }
+
+ static int toCheckedInt(Number n) {
+ BigInteger bi = toBigIntegerTruncated(n);
+ if (bi != null) {
+ if (bi.compareTo(BigInteger.valueOf(Integer.MAX_VALUE)) > 0
+ || bi.compareTo(BigInteger.valueOf(Integer.MIN_VALUE)) <
0) {
+ throw new TableRuntimeException("Value " + n + " is out of
range for INTEGER");
+ }
+ return bi.intValue();
+ }
+ long v = n.longValue();
+ if (v < Integer.MIN_VALUE || v > Integer.MAX_VALUE) {
+ throw new TableRuntimeException("Value " + n + " is out of range
for INTEGER");
+ }
+ return (int) v;
+ }
+
+ static float toCheckedFloat(Number n) {
+ float f = n.floatValue();
+ if (Float.isInfinite(f) || Float.isNaN(f)) {
+ throw new TableRuntimeException("Value " + n + " is out of range
for FLOAT");
+ }
+ return f;
+ }
+
+ static double toCheckedDouble(Number n) {
+ double d = n.doubleValue();
+ if (Double.isInfinite(d) || Double.isNaN(d)) {
+ throw new TableRuntimeException("Value " + n + " is out of range
for DOUBLE");
+ }
+ return d;
+ }
+
+ static long toCheckedLong(Number n) {
+ BigInteger bi = toBigIntegerTruncated(n);
+ if (bi != null) {
+ if (bi.compareTo(BigInteger.valueOf(Long.MAX_VALUE)) > 0
+ || bi.compareTo(BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
+ throw new TableRuntimeException("Value " + n + " is out of
range for BIGINT");
+ }
+ return bi.longValue();
+ }
+ return n.longValue();
+ }
+
+ private static Number asNumber(Object raw, LogicalTypeRoot typeRoot) {
+ if (!(raw instanceof Number)) {
+ throw new TableRuntimeException(
+ "Cannot convert " + raw.getClass().getName() + " to " +
typeRoot);
+ }
+ return (Number) raw;
+ }
+
+ static DecimalData toCheckedDecimal(Object raw, int precision, int scale) {
+ if (!(raw instanceof Number)) {
+ throw new TableRuntimeException(
+ "Cannot convert " + raw.getClass().getName() + " to
DECIMAL");
+ }
+ if (raw instanceof BigDecimal) {
+ final BigDecimal bd = (BigDecimal) raw;
+ final long integerDigits = (long) bd.precision() - bd.scale();
+ if (integerDigits <= -(long) scale - 1) {
+ return DecimalData.fromBigDecimal(BigDecimal.ZERO, precision,
scale);
+ }
+ if (integerDigits > precision - scale) {
+ throw new TableRuntimeException(
+ "Value "
+ + raw
+ + " cannot be represented as DECIMAL("
+ + precision
+ + ", "
+ + scale
+ + ")");
+ }
+ }
+ DecimalData result = DecimalDataUtils.castFrom(raw.toString(),
precision, scale);
+ if (result == null) {
+ throw new TableRuntimeException(
+ "Value "
+ + raw
+ + " cannot be represented as DECIMAL("
+ + precision
+ + ", "
+ + scale
+ + ")");
+ }
+ return result;
+ }
+
public static class JsonValueContext {
@JsonValue public final Object obj;
public final Exception exc;
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/SqlJsonUtilsConversionTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/SqlJsonUtilsConversionTest.java
new file mode 100644
index 00000000000..2ca162680ec
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/SqlJsonUtilsConversionTest.java
@@ -0,0 +1,559 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.table.api.JsonValueOnEmptyOrError;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.JsonQueryOnEmptyOrError.EMPTY_ARRAY;
+import static org.apache.flink.table.api.JsonQueryOnEmptyOrError.ERROR;
+import static org.apache.flink.table.api.JsonQueryOnEmptyOrError.NULL;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.BIGINT;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.BOOLEAN;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.DECIMAL;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.DOUBLE;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.FLOAT;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTEGER;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.SMALLINT;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.TINYINT;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SqlJsonUtilsConversionTest {
+
+ private static final BigInteger HUGE = new
BigInteger("18446744073709551616");
+
+ private static Object scalar(Object raw, LogicalTypeRoot type) {
+ return SqlJsonUtils.convertJsonScalar(raw, type, 0, 0,
JsonValueOnEmptyOrError.NULL, null);
+ }
+
+ private static Object scalar(
+ Object raw,
+ LogicalTypeRoot type,
+ JsonValueOnEmptyOrError onError,
+ Object defaultValue) {
+ return SqlJsonUtils.convertJsonScalar(raw, type, 0, 0, onError,
defaultValue);
+ }
+
+ private static Object scalarDecimal(Object raw, int precision, int scale) {
+ return SqlJsonUtils.convertJsonScalar(
+ raw, DECIMAL, precision, scale, JsonValueOnEmptyOrError.NULL,
null);
+ }
+
+ private static GenericArrayData array(Object[] raw, LogicalTypeRoot
elementType) {
+ return SqlJsonUtils.convertJsonArray(raw, elementType, 0, 0, true,
NULL);
+ }
+
+ private static GenericArrayData arrayDecimal(
+ Object[] raw,
+ int precision,
+ int scale,
+ org.apache.flink.table.api.JsonQueryOnEmptyOrError onError) {
+ return SqlJsonUtils.convertJsonArray(raw, DECIMAL, precision, scale,
true, onError);
+ }
+
+ // ------------------------------------------------------------------------
+ // Numeric range conversion
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> numericValueConversions() {
+ return Stream.of(
+ Arguments.of(42, TINYINT, (byte) 42),
+ Arguments.of(127L, TINYINT, Byte.MAX_VALUE),
+ Arguments.of(-128L, TINYINT, Byte.MIN_VALUE),
+ Arguments.of(1000, SMALLINT, (short) 1000),
+ Arguments.of((long) Short.MAX_VALUE, SMALLINT,
Short.MAX_VALUE),
+ Arguments.of(42L, INTEGER, 42),
+ Arguments.of((long) Integer.MAX_VALUE, INTEGER,
Integer.MAX_VALUE),
+ Arguments.of(42, BIGINT, 42L),
+ Arguments.of(9_999_999_999L, BIGINT, 9_999_999_999L),
+ Arguments.of(new BigDecimal("9223372036854775807.5"), BIGINT,
Long.MAX_VALUE),
+ Arguments.of(13.37, FLOAT, 13.37f),
+ Arguments.of(13.37, DOUBLE, 13.37),
+ // fraction truncation
+ Arguments.of(13.37, INTEGER, 13),
+ Arguments.of(-13.99, INTEGER, -13),
+ Arguments.of(new BigDecimal("9999999999.5"), BIGINT,
9_999_999_999L),
+ Arguments.of(new BigDecimal("-42.9"), BIGINT, -42L),
+ // silent precision loss when the target type cannot hold the
value exactly
+ Arguments.of(16777217, FLOAT, 16777216.0f),
+ Arguments.of(9007199254740993L, DOUBLE, 9.007199254740992E15));
+ }
+
+ static Stream<Arguments> numericOverflowCases() {
+ return Stream.of(
+ Arguments.of(128, TINYINT),
+ Arguments.of(-129, TINYINT),
+ Arguments.of(200L, TINYINT),
+ Arguments.of(-200L, TINYINT),
+ Arguments.of(HUGE, TINYINT),
+ Arguments.of(HUGE.negate(), TINYINT),
+ Arguments.of(new BigDecimal("18446744073709551616"), TINYINT),
+ Arguments.of((long) Short.MAX_VALUE + 1, SMALLINT),
+ Arguments.of((long) Short.MIN_VALUE - 1, SMALLINT),
+ Arguments.of(40_000L, SMALLINT),
+ Arguments.of(-40_000L, SMALLINT),
+ Arguments.of(HUGE, SMALLINT),
+ Arguments.of(HUGE.negate(), SMALLINT),
+ Arguments.of(new BigDecimal("18446744073709551616"), SMALLINT),
+ Arguments.of(9_999_999_999L, INTEGER),
+ Arguments.of(-9_999_999_999L, INTEGER),
+ Arguments.of(HUGE, INTEGER),
+ Arguments.of(HUGE.negate(), INTEGER),
+ Arguments.of(new BigDecimal("18446744073709551616"), INTEGER),
+ Arguments.of(HUGE, BIGINT),
+ Arguments.of(HUGE.negate(), BIGINT),
+ Arguments.of(new BigDecimal("18446744073709551616"), BIGINT),
+
Arguments.of(BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.ONE), BIGINT),
+
Arguments.of(BigInteger.valueOf(Long.MIN_VALUE).subtract(BigInteger.ONE),
BIGINT),
+ Arguments.of(new
BigDecimal(Long.MAX_VALUE).add(BigDecimal.ONE), BIGINT),
+ Arguments.of(1e40, FLOAT),
+ Arguments.of(-1e40, FLOAT),
+ Arguments.of(new BigDecimal("1E309"), DOUBLE),
+ Arguments.of(new BigDecimal("-1E309"), DOUBLE),
+ // Double/Float inputs to integer types must range-check, not
saturate via
+ // Number.xxxValue() ((long) 1e20 -> Long.MAX_VALUE, (long)
NaN -> 0)
+ Arguments.of(1e20, TINYINT),
+ Arguments.of(1e20, SMALLINT),
+ Arguments.of(1e20, INTEGER),
+ Arguments.of(1e20, BIGINT),
+ Arguments.of(1e20f, BIGINT),
+ Arguments.of(Double.NaN, BIGINT),
+ Arguments.of(Double.POSITIVE_INFINITY, BIGINT),
+ // huge exponent: reject before toBigInteger() materializes
10^999999999
+ Arguments.of(new BigDecimal("1E999999999"), INTEGER),
+ Arguments.of(new BigDecimal("1E999999999"), BIGINT));
+ }
+
+ @Nested
+ class NumericRangeConversion {
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#numericValueConversions")
+ void convertsToExpectedValue(Number input, LogicalTypeRoot type,
Object expected) {
+ assertThat(scalar(input, type)).isEqualTo(expected);
+ }
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#numericOverflowCases")
+ void overflowReturnsNull(Number input, LogicalTypeRoot type) {
+ assertThat(scalar(input, type)).isNull();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Decimal conversion
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> decimalConversions() {
+ return Stream.of(
+ Arguments.of(13.37, 10, 2, "13.37"),
+ Arguments.of(
+ "99999999999999999999999999999999999999",
+ 38,
+ 0,
+ "99999999999999999999999999999999999999"),
+ Arguments.of("123.456", 5, 2, "123.46"),
+ // huge positive scale: round to zero before setScale()
materializes 10^999999999
+ Arguments.of(new BigDecimal("1E-999999999"), 10, 2, "0"));
+ }
+
+ static Stream<Arguments> decimalOverflowCases() {
+ return Stream.of(
+ Arguments.of(123456789.99, 5, 2),
+ Arguments.of("999999999999999999999999999999999999999", 38, 0),
+ // 999.999 rounds to 1000.00 (precision 6) which overflows
DECIMAL(5,2)
+ Arguments.of("999.999", 5, 2),
+ // huge exponent: reject before setScale() materializes
10^999999999
+ Arguments.of(new BigDecimal("1E999999999"), 10, 2));
+ }
+
+ @Nested
+ class DecimalConversion {
+
+ @ParameterizedTest(name = "{0} -> DECIMAL({1},{2})")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#decimalConversions")
+ void convertsToDecimal(Object raw, int precision, int scale, String
expected) {
+ Object result = scalarDecimal(raw, precision, scale);
+ assertThat(result).isInstanceOf(DecimalData.class);
+ assertThat(((DecimalData) result).toBigDecimal())
+ .isEqualByComparingTo(new BigDecimal(expected));
+ }
+
+ @ParameterizedTest(name = "{0} -> DECIMAL({1},{2})")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#decimalOverflowCases")
+ void overflowReturnsNull(Object raw, int precision, int scale) {
+ assertThat(scalarDecimal(raw, precision, scale)).isNull();
+ }
+
+ @Test
+ void scientificNotation() {
+ Object result = scalarDecimal("1.5E1", 10, 2);
+ assertThat(result).isInstanceOf(DecimalData.class);
+ assertThat(((DecimalData) result).toBigDecimal())
+ .isEqualByComparingTo(new BigDecimal("15.00"));
+ }
+
+ @Test
+ void precisionOverflowWithStringDefault() {
+ Object result =
+ SqlJsonUtils.convertJsonScalar(
+ 123456789.99, DECIMAL, 5, 2,
JsonValueOnEmptyOrError.DEFAULT, "0.00");
+ assertThat(result).isInstanceOf(DecimalData.class);
+ assertThat(((DecimalData) result).toBigDecimal())
+ .isEqualByComparingTo(new BigDecimal("0.00"));
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Boolean conversion
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> booleanConversions() {
+ return Stream.of(
+ Arguments.of(0, false),
+ Arguments.of(1, true),
+ Arguments.of(0L, false),
+ Arguments.of(1L, true),
+ Arguments.of(0.0, false),
+ Arguments.of(1.0, true),
+ Arguments.of("true", true),
+ Arguments.of("false", false),
+ Arguments.of("TRUE", true),
+ Arguments.of("t", true),
+ Arguments.of("f", false),
+ Arguments.of("yes", true),
+ Arguments.of("no", false),
+ Arguments.of("y", true),
+ Arguments.of("n", false),
+ Arguments.of("Y", true),
+ Arguments.of("N", false),
+ Arguments.of("1", true),
+ Arguments.of("0", false),
+ // leading/trailing whitespace is trimmed before matching
+ Arguments.of(" true ", true),
+ Arguments.of(" FALSE ", false));
+ }
+
+ static Stream<Arguments> nonBooleanInputs() {
+ return Stream.of(
+ Arguments.of(42, "Cannot convert 42 to BOOLEAN"),
+ Arguments.of(-1, "Cannot convert -1 to BOOLEAN"),
+ Arguments.of(2, "Cannot convert 2 to BOOLEAN"),
+ Arguments.of(1.9, "Cannot convert 1.9 to BOOLEAN"),
+ Arguments.of(0.5, "Cannot convert 0.5 to BOOLEAN"),
+ Arguments.of(-0.1, "Cannot convert -0.1 to BOOLEAN"),
+ // exact BigDecimal comparison rejects values that
doubleValue() would round to 0/1
+ Arguments.of(
+ new BigDecimal("1.00000000000000000001"),
+ "Cannot convert 1.00000000000000000001 to BOOLEAN"),
+ Arguments.of(
+ new BigDecimal("0.99999999999999999999"),
+ "Cannot convert 0.99999999999999999999 to BOOLEAN"),
+ Arguments.of(new BigDecimal("1E-400"), "Cannot convert 1E-400
to BOOLEAN"),
+ Arguments.of("maybe", "Cannot cast 'maybe' as BOOLEAN."),
+ Arguments.of("2", "Cannot cast '2' as BOOLEAN."));
+ }
+
+ @Nested
+ class BooleanConversion {
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#booleanConversions")
+ void convertsToBoolean(Object raw, Boolean expected) {
+ assertThat(scalar(raw, BOOLEAN)).isEqualTo(expected);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#nonBooleanInputs")
+ void failsForNonBoolean(Object raw, String causeMessage) {
+ assertThatThrownBy(() -> scalar(raw, BOOLEAN,
JsonValueOnEmptyOrError.ERROR, null))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Cannot cast " + raw.getClass().getName() + "
to BOOLEAN")
+ .cause()
+ .hasMessage(causeMessage);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // String input parsing
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> stringConversions() {
+ return Stream.of(
+ Arguments.of("42", INTEGER, 42),
+ Arguments.of("13.37", FLOAT, 13.37f),
+ Arguments.of("9999999999", BIGINT, 9_999_999_999L),
+ // CAST trims leading/trailing whitespace (matches PostgreSQL)
+ Arguments.of(" 42 ", INTEGER, 42),
+ Arguments.of(" 13.37 ", DOUBLE, 13.37));
+ }
+
+ static Stream<Arguments> invalidStringInputs() {
+ return Stream.of(
+ Arguments.of("9999999999", INTEGER),
+ Arguments.of("", INTEGER),
+ Arguments.of("NaN", FLOAT),
+ Arguments.of("NaN", DOUBLE),
+ Arguments.of("NaN", INTEGER),
+ Arguments.of("Infinity", FLOAT),
+ Arguments.of("-Infinity", DOUBLE));
+ }
+
+ @Nested
+ class StringInputParsing {
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#stringConversions")
+ void parsesToExpectedValue(String raw, LogicalTypeRoot type, Object
expected) {
+ assertThat(scalar(raw, type)).isEqualTo(expected);
+ }
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#invalidStringInputs")
+ void returnsNull(String raw, LogicalTypeRoot type) {
+ assertThat(scalar(raw, type)).isNull();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // On-error behavior
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> onErrorNullInputs() {
+ return Stream.of(
+ // null input passes straight through
+ Arguments.of(null, INTEGER),
+ // a boolean cannot be converted to a number
+ Arguments.of(Boolean.TRUE, INTEGER));
+ }
+
+ @Nested
+ class OnErrorBehavior {
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#onErrorNullInputs")
+ void returnsNull(Object raw, LogicalTypeRoot type) {
+ assertThat(scalar(raw, type)).isNull();
+ }
+
+ @Test
+ void returnsDefaultForUnparsableString() {
+ assertThat(scalar("notANumber", INTEGER,
JsonValueOnEmptyOrError.DEFAULT, 42))
+ .isEqualTo(42);
+ }
+
+ @Test
+ void throwsOnErrorMode() {
+ assertThatThrownBy(
+ () ->
+ scalar(
+ 9_999_999_999L,
+ INTEGER,
+ JsonValueOnEmptyOrError.ERROR,
+ null))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Cannot cast java.lang.Long to INTEGER")
+ .cause()
+ .hasMessage("Value 9999999999 is out of range for
INTEGER");
+ }
+
+ @Test
+ void errorModeSurfacesParseCause() {
+ assertThatThrownBy(() -> scalar("abc", INTEGER,
JsonValueOnEmptyOrError.ERROR, null))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Cannot cast java.lang.String to INTEGER")
+ .cause()
+ .hasMessage("Cannot parse string 'abc' as INTEGER");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Array conversion
+ // ------------------------------------------------------------------------
+
+ static Stream<Arguments> arrayConversions() {
+ return Stream.of(
+ Arguments.of(new Object[] {1, 2, 3}, INTEGER, new Object[] {1,
2, 3}),
+ Arguments.of(new Object[] {1, null, 3}, INTEGER, new Object[]
{1, null, 3}),
+ Arguments.of(new Object[] {true, false}, BOOLEAN, new Object[]
{true, false}),
+ Arguments.of(new Object[] {0, 1}, BOOLEAN, new Object[]
{false, true}),
+ Arguments.of(new Object[] {"1", "2", "3"}, INTEGER, new
Object[] {1, 2, 3}),
+ Arguments.of(new Object[] {"yes", "no"}, BOOLEAN, new Object[]
{true, false}),
+ Arguments.of(new Object[] {"Y", "n"}, BOOLEAN, new Object[]
{true, false}),
+ Arguments.of(
+ new Object[] {1, 9999999999L}, BIGINT, new Object[]
{1L, 9999999999L}));
+ }
+
+ static Stream<Arguments> arrayNullResultCases() {
+ return Stream.of(
+ Arguments.of(new Object[] {"a", "b"}, INTEGER, 0, 0, true),
+ Arguments.of(new Object[] {1, "notANumber", 3}, INTEGER, 0, 0,
true),
+ // element 2 is neither 0 nor 1, so conversion fails atomically
+ Arguments.of(new Object[] {1, 2, 3}, BOOLEAN, 0, 0, true),
+ Arguments.of(new Object[] {999999}, DECIMAL, 2, 1, true),
+ // 999.999 rounds to 1000.00 which overflows DECIMAL(5,2);
whole array fails
+ Arguments.of(new Object[] {1.5, 999.999}, DECIMAL, 5, 2, true),
+ // a null element in a NOT NULL array fails
+ Arguments.of(new Object[] {1, null, 3}, INTEGER, 0, 0, false));
+ }
+
+ static Stream<Arguments> arrayEmptyResultCases() {
+ return Stream.of(
+ Arguments.of(new Object[] {"a", "b"}, INTEGER, 0, 0),
+ Arguments.of(new Object[] {999999}, DECIMAL, 2, 1));
+ }
+
+ static Stream<Arguments> arrayThrowCases() {
+ return Stream.of(
+ Arguments.of(
+ new Object[] {"a", "b"},
+ INTEGER,
+ 0,
+ 0,
+ true,
+ "Cannot parse string 'a' as INTEGER"),
+ Arguments.of(
+ new Object[] {999999},
+ DECIMAL,
+ 2,
+ 1,
+ true,
+ "Value 999999 cannot be represented as DECIMAL(2, 1)"),
+ // a null element in a NOT NULL array must surface the real
cause through the chain
+ Arguments.of(
+ new Object[] {1, null, 3},
+ INTEGER,
+ 0,
+ 0,
+ false,
+ "Null element at index 1 in NOT NULL array"));
+ }
+
+ static Stream<Arguments> emptyArrayInputs() {
+ return Stream.of(Arguments.of(INTEGER, 0, 0), Arguments.of(DECIMAL,
10, 2));
+ }
+
+ @Nested
+ class ArrayConversion {
+
+ @ParameterizedTest(name = "{1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#arrayConversions")
+ void convertsArray(Object[] raw, LogicalTypeRoot elementType, Object[]
expected) {
+ assertThat(array(raw,
elementType).toObjectArray()).containsExactly(expected);
+ }
+
+ @Test
+ void convertsDecimalArrayWithRounding() {
+ GenericArrayData result = arrayDecimal(new Object[] {1.5,
123.456}, 5, 2, NULL);
+ assertThat(result).isNotNull();
+ Object[] elements = result.toObjectArray();
+ assertThat(((DecimalData) elements[0]).toBigDecimal())
+ .isEqualByComparingTo(new BigDecimal("1.50"));
+ assertThat(((DecimalData) elements[1]).toBigDecimal())
+ .isEqualByComparingTo(new BigDecimal("123.46"));
+ }
+
+ @ParameterizedTest(name = "{1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#arrayNullResultCases")
+ void returnsNullOnErrorMode(
+ Object[] raw,
+ LogicalTypeRoot elementType,
+ int precision,
+ int scale,
+ boolean nullable) {
+ assertThat(
+ SqlJsonUtils.convertJsonArray(
+ raw, elementType, precision, scale,
nullable, NULL))
+ .isNull();
+ }
+
+ @ParameterizedTest(name = "{1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#arrayEmptyResultCases")
+ void returnsEmptyArrayOnErrorMode(
+ Object[] raw, LogicalTypeRoot elementType, int precision, int
scale) {
+ GenericArrayData result =
+ SqlJsonUtils.convertJsonArray(
+ raw, elementType, precision, scale, true,
EMPTY_ARRAY);
+ assertThat(result).isNotNull();
+ assertThat(result.size()).isZero();
+ }
+
+ @ParameterizedTest(name = "{1}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#arrayThrowCases")
+ void throwsOnErrorMode(
+ Object[] raw,
+ LogicalTypeRoot elementType,
+ int precision,
+ int scale,
+ boolean nullable,
+ String causeMessage) {
+ assertThatThrownBy(
+ () ->
+ SqlJsonUtils.convertJsonArray(
+ raw, elementType, precision,
scale, nullable, ERROR))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("JSON_QUERY array element conversion failed")
+ .cause()
+ .hasMessage(causeMessage);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource(
+
"org.apache.flink.table.runtime.functions.SqlJsonUtilsConversionTest#emptyArrayInputs")
+ void emptyInputStaysEmpty(LogicalTypeRoot elementType, int precision,
int scale) {
+ GenericArrayData result =
+ SqlJsonUtils.convertJsonArray(
+ new Object[0], elementType, precision, scale,
true, NULL);
+ assertThat(result).isNotNull();
+ assertThat(result.size()).isZero();
+ }
+
+ @Test
+ void nullInputPassesThrough() {
+ assertThat(SqlJsonUtils.convertJsonArray(null, INTEGER, 0, 0,
true, NULL)).isNull();
+ }
+ }
+}
