cjohnson-confluent commented on code in PR #29063:
URL: https://github.com/apache/flink/pull/29063#discussion_r3936294140
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java:
##########
@@ -815,6 +828,233 @@ public String toString() {
}
}
+ // --- Type conversion for JSON_VALUE / JSON_QUERY RETURNING ---
+
+ private static final java.util.EnumSet<LogicalTypeRoot>
SUPPORTED_JSON_RETURNING_TYPES =
+ java.util.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 (JsonConversionException 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,
+ 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);
+ }
+ }
+ return new GenericArrayData(converted);
+ } catch (JsonConversionException e) {
+ switch (errorBehavior) {
+ case NULL:
+ return null;
+ case EMPTY_ARRAY:
+ return new GenericArrayData(new Object[0]);
+ case ERROR:
+ throw new TableRuntimeException("Array element type
mismatch in JSON_QUERY", e);
+ default:
+ return null;
+ }
+ }
+ }
+
+ private static Object convertToType(
+ Object raw, LogicalTypeRoot typeRoot, int precision, int scale) {
+ if (raw instanceof StringData) {
+ return convertToType(raw.toString(), typeRoot, precision, scale);
+ }
+ if (raw instanceof String) {
+ if (typeRoot == LogicalTypeRoot.BOOLEAN) {
+ return parseStringAsBoolean((String) raw);
+ }
+ try {
+ return convertToType(new BigDecimal((String) raw), typeRoot,
precision, scale);
+ } catch (NumberFormatException e) {
+ throw new JsonConversionException(
+ "Cannot parse string '" + raw + "' as " + typeRoot, e);
+ }
+ }
+ try {
+ switch (typeRoot) {
+ case BOOLEAN:
+ return (Boolean) raw;
Review Comment:
Changed the BOOLEAN case to explicitly handle Number inputs (0/1
conversion), with the `(Boolean)` cast remaining as a fallback for actual
Boolean values. The cast still guards against unexpected types like Map or List.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java:
##########
@@ -815,6 +828,233 @@ public String toString() {
}
}
+ // --- Type conversion for JSON_VALUE / JSON_QUERY RETURNING ---
+
+ private static final java.util.EnumSet<LogicalTypeRoot>
SUPPORTED_JSON_RETURNING_TYPES =
+ java.util.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 (JsonConversionException 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,
+ 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);
+ }
+ }
+ return new GenericArrayData(converted);
+ } catch (JsonConversionException e) {
+ switch (errorBehavior) {
+ case NULL:
+ return null;
+ case EMPTY_ARRAY:
+ return new GenericArrayData(new Object[0]);
+ case ERROR:
+ throw new TableRuntimeException("Array element type
mismatch in JSON_QUERY", e);
+ default:
+ return null;
+ }
+ }
+ }
+
+ private static Object convertToType(
+ Object raw, LogicalTypeRoot typeRoot, int precision, int scale) {
+ if (raw instanceof StringData) {
+ return convertToType(raw.toString(), typeRoot, precision, scale);
+ }
+ if (raw instanceof String) {
+ if (typeRoot == LogicalTypeRoot.BOOLEAN) {
+ return parseStringAsBoolean((String) raw);
+ }
+ try {
+ return convertToType(new BigDecimal((String) raw), typeRoot,
precision, scale);
+ } catch (NumberFormatException e) {
+ throw new JsonConversionException(
+ "Cannot parse string '" + raw + "' as " + typeRoot, e);
+ }
+ }
+ try {
+ switch (typeRoot) {
+ case BOOLEAN:
+ return (Boolean) raw;
+ case TINYINT:
+ return toCheckedByte((Number) raw);
+ case SMALLINT:
+ return toCheckedShort((Number) raw);
+ case INTEGER:
+ return toCheckedInt((Number) raw);
+ case BIGINT:
+ return toCheckedLong((Number) raw);
+ case FLOAT:
+ return toCheckedFloat((Number) raw);
+ case DOUBLE:
+ return toCheckedDouble((Number) raw);
+ case DECIMAL:
+ return toCheckedDecimal(((Number) raw).toString(),
precision, scale);
Review Comment:
Applied.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -528,6 +567,159 @@ private static TestSetSpec jsonValueSpec() {
"JSON_VALUE(f0, 'strict $.invalid' RETURNING INTEGER
NULL ON EMPTY DEFAULT 42 ON ERROR)",
42,
INT())
+ .testResult(
+ $("f0").jsonValue(
+ "lax $.invalid",
+ BIGINT(),
+ JsonValueOnEmptyOrError.DEFAULT,
+ 99L,
+ JsonValueOnEmptyOrError.ERROR,
+ null),
+ "JSON_VALUE(f0, 'lax $.invalid' RETURNING BIGINT
DEFAULT 99 ON EMPTY ERROR ON ERROR)",
+ 99L,
+ BIGINT())
+
+ // 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())
+ .testResult(
+ $("f0").jsonValue(
+ "$.type",
+ INT(),
+ JsonValueOnEmptyOrError.NULL,
+ null,
+ JsonValueOnEmptyOrError.NULL,
+ null),
+ "JSON_VALUE(f0, '$.type' RETURNING INTEGER NULL ON
ERROR)",
+ null,
+ INT())
+ .testResult(
+ $("f0").jsonValue(
+ "$.type",
+ BIGINT(),
+ JsonValueOnEmptyOrError.NULL,
+ null,
+ JsonValueOnEmptyOrError.DEFAULT,
+ 0L),
+ "JSON_VALUE(f0, '$.type' RETURNING BIGINT DEFAULT 0 ON
ERROR)",
+ 0L,
+ BIGINT())
+ .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 (13.37 -> 13): MySQL truncates,
PostgreSQL errors.
+ // We match MySQL (CAST(JSON_UNQUOTE(JSON_EXTRACT(...)) AS
type)).
Review Comment:
Updated: "Fractional truncation toward zero (13.89 -> 13, -13.89 -> -13)".
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -886,7 +1083,214 @@ private static List<TestSetSpec> jsonQuerySpec() {
.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]}")
Review Comment:
Not supported. Typed RETURNING is a Flink extension (standard SQL JSON_QUERY
returns VARCHAR), and MAP is out of scope for this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
