cjohnson-confluent commented on code in PR #29063:
URL: https://github.com/apache/flink/pull/29063#discussion_r3936292684
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -480,6 +484,41 @@ private static TestSetSpec jsonValueSpec() {
"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 java.math.BigDecimal("13.37"),
Review Comment:
Added `import java.math.BigDecimal` and shortened all occurrences.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -480,6 +484,41 @@ private static TestSetSpec jsonValueSpec() {
"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 java.math.BigDecimal("13.37"),
+ DECIMAL(10, 2))
+ .testResult(
+ $("f0").jsonValue("$.longBalance", DECIMAL(30, 10)),
Review Comment:
Added.
##########
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);
+ default:
+ throw new JsonConversionException(
+ "Unsupported type for JSON conversion: " +
typeRoot);
+ }
+ } catch (ClassCastException e) {
+ throw new JsonConversionException(
+ "Cannot convert " + raw.getClass().getName() + " to " +
typeRoot, e);
+ }
+ }
+
+ 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 (JsonConversionException e) {
+ throw new TableRuntimeException(
+ "Default value " + defaultValue + " cannot be represented
as " + typeRoot, e);
+ }
+ }
+
+ static byte toCheckedByte(Number n) {
+ long v = n.longValue();
+ if (v < Byte.MIN_VALUE || v > Byte.MAX_VALUE) {
+ throw new JsonConversionException("Value " + n + " is out of range
for TINYINT");
+ }
+ return (byte) v;
+ }
+
+ static short toCheckedShort(Number n) {
+ long v = n.longValue();
+ if (v < Short.MIN_VALUE || v > Short.MAX_VALUE) {
+ throw new JsonConversionException("Value " + n + " is out of range
for SMALLINT");
+ }
+ return (short) v;
+ }
+
+ static int toCheckedInt(Number n) {
+ long v = n.longValue();
Review Comment:
Good catch. `toCheckedByte`/`toCheckedShort`/`toCheckedInt` were calling
`n.longValue()` before range-checking, which silently truncates BigInteger
values beyond long range. I extracted a `toBigIntegerTruncated()` helper
(shared with `toCheckedLong`) that guards against this. Unit tests added for
`BigInteger(2^64)` and `BigDecimal` equivalents across all integer types.
##########
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);
Review Comment:
Now supported. Integer `0` -> `false`, `1` -> `true`, other integers trigger
ON ERROR. Consistent with string conversion where `"0"` and `"1"` already map
to boolean via `parseStringAsBoolean`. Works for both JSON_VALUE and JSON_QUERY
RETURNING ARRAY<BOOLEAN>.
--
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]
