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 34c36e63714 [FLINK-41087][table] Don't build the `PARSE_JSON` error
for rows `TRY_PARSE_JSON` rejects (#29579)
34c36e63714 is described below
commit 34c36e6371467181bd0a130b760b5fd681d608dc
Author: Ramin Gharib <[email protected]>
AuthorDate: Sun Oct 25 17:43:34 2026 +0100
[FLINK-41087][table] Don't build the `PARSE_JSON` error for rows
`TRY_PARSE_JSON` rejects (#29579)
PARSE_JSON and TRY_PARSE_JSON share one generated parse per input. The
shared parse called SqlJsonUtils#parseJsonToVariant, which scanned the input
for invalid UTF-8, formatted it into a message and created a
TableRuntimeException on every failure. TRY_PARSE_JSON then dropped that
exception and returned NULL, so every row it rejected paid for a second
exception with its stack trace and a copy of the input.
parseJsonToVariant now lets the parser's exception through. The new
SqlJsonUtils#parseJsonError builds the TableRuntimeException with the UTF-8
diagnostics, and only a PARSE_JSON call site calls it. Error messages and
results do not change.
Generated-by: Claude Code 2.1.309 (Claude Opus 5.5)
---
.../table/planner/codegen/JsonCodeGenUtils.java | 14 +++++--
.../planner/codegen/calls/BuiltInMethods.scala | 6 +++
.../table/planner/codegen/JsonParseReuseTest.java | 33 +++++++++++++++--
.../table/runtime/functions/SqlJsonUtils.java | 43 ++++++++++++++--------
4 files changed, 72 insertions(+), 24 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
index 2dd2f068c71..8ab8c83580b 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
@@ -43,7 +43,9 @@ public final class JsonCodeGenUtils {
*
* <p>The input is parsed at most once per record for a given (input,
allowDuplicateKeys) pair
* and shared across all such calls; {@code PARSE_JSON} rethrows a parse
error while {@code
- * TRY_PARSE_JSON} returns NULL.
+ * TRY_PARSE_JSON} returns NULL. The shared parse keeps the parser's
exception as is. Only a
+ * {@code PARSE_JSON} call site turns it into the user-facing error, so a
row that {@code
+ * TRY_PARSE_JSON} rejects costs no more than the failed parse.
*
* <p>For {@code PARSE_JSON(str)} the generated code looks like:
*
@@ -66,7 +68,7 @@ public final class JsonCodeGenUtils {
* SqlJsonUtils.parseJsonToVariant(in,
java.lang.Boolean.TRUE.equals(allowDup));
* } catch (Exception e) {
* variantParsed$0 = null;
- * variantParseError$0 = t;
+ * variantParseError$0 = e;
* }
* return variantParsed$0;
* }
@@ -77,7 +79,7 @@ public final class JsonCodeGenUtils {
* if (!inIsNull) {
* result$0 = parseVariant$0(in, null);
* if (variantParseError$0 != null) { // only emitted for PARSE_JSON,
not TRY_PARSE_JSON
- * org.apache.flink.util.ExceptionUtils.rethrow(variantParseError$0);
+ * throw SqlJsonUtils.parseJsonError(in, variantParseError$0);
* }
* isNull$0 = result$0 == null;
* }
@@ -207,7 +209,11 @@ public final class JsonCodeGenUtils {
: " if ("
+ errorTerm
+ " != null) {\n"
- + "
org.apache.flink.util.ExceptionUtils.rethrow("
+ + " throw "
+ +
CodeGenUtils.qualifyMethod(BuiltInMethods.PARSE_JSON_ERROR())
+ + "("
+ + input.resultTerm()
+ + ", "
+ errorTerm
+ ");\n"
+ " }\n";
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
index 9bb1de380a5..e1d78cae339 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -520,6 +520,12 @@ object BuiltInMethods {
classOf[StringData],
classOf[Boolean])
+ val PARSE_JSON_ERROR = Types.lookupMethod(
+ classOf[SqlJsonUtils],
+ "parseJsonError",
+ classOf[StringData],
+ classOf[Throwable])
+
val JSON_TYPE =
Types.lookupMethod(classOf[SqlJsonUtils], "jsonType",
classOf[SqlJsonUtils.JsonValueContext])
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
index a429e541911..b6f4e63189e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
@@ -53,6 +53,10 @@ class JsonParseReuseTest {
private static final Pattern PARSE_JSON_PATTERN =
Pattern.compile("\\b" +
Pattern.quote(BuiltInMethods.PARSE_JSON().getName()) + "\\(");
+ private static final Pattern PARSE_JSON_ERROR_PATTERN =
+ Pattern.compile(
+ "\\b" +
Pattern.quote(BuiltInMethods.PARSE_JSON_ERROR().getName()) + "\\(");
+
private StreamTableEnvironment tEnv;
private static final String JSON_ROW1 =
@@ -84,6 +88,10 @@ class JsonParseReuseTest {
return count(PARSE_JSON_PATTERN, code);
}
+ private static int countParseJsonError(final String code) {
+ return count(PARSE_JSON_ERROR_PATTERN, code);
+ }
+
private static int count(final Pattern pattern, final String code) {
final Matcher m = pattern.matcher(code);
int count = 0;
@@ -529,14 +537,31 @@ class JsonParseReuseTest {
final String badSrc = "(VALUES ('not json')) AS t(j)";
final String sharedSql =
"SELECT JSON_STRING(TRY_PARSE_JSON(j)),
JSON_STRING(PARSE_JSON(j)) FROM " + badSrc;
- assertThat(countParseJson(extractGeneratedCode(sharedSql)))
+ final String sharedCode = extractGeneratedCode(sharedSql);
+ assertThat(countParseJson(sharedCode))
.as("PARSE_JSON + TRY_PARSE_JSON on the same input should
parse once")
.isOne();
+ assertThat(countParseJsonError(sharedCode))
+ .as("Only the PARSE_JSON call site builds the parse error")
+ .isOne();
assertThatThrownBy(() -> collect(sharedSql))
- .hasStackTraceContaining("Failed to parse JSON string");
+ .hasStackTraceContaining("Failed to parse JSON string: not
json");
+ }
- assertThat(collect("SELECT JSON_STRING(TRY_PARSE_JSON(j)) FROM " +
badSrc))
- .containsExactly(Row.of((String) null));
+ @Test
+ void testTryParseJsonDoesNotBuildParseError() {
+ tEnv.createTemporaryView(
+ "mixed_src", tEnv.fromValues(Row.of("not json"),
Row.of("{\"a\":1}")).as("j"));
+ final String sql = "SELECT JSON_STRING(TRY_PARSE_JSON(j)) FROM
mixed_src";
+ assertThat(collect(sql))
+ .containsExactlyInAnyOrder(Row.of((String) null),
Row.of("{\"a\":1}"));
+ final String code = extractGeneratedCode(sql);
+ assertThat(countParseJson(code))
+ .as("TRY_PARSE_JSON must be generated, not constant-folded
away")
+ .isOne();
+ assertThat(countParseJsonError(code))
+ .as("TRY_PARSE_JSON drops the parser's exception without
building the error")
+ .isZero();
}
@Test
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 b319d519e73..431d0248cf6 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
@@ -58,6 +58,7 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.Obje
import javax.annotation.Nullable;
+import java.io.IOException;
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.math.BigInteger;
@@ -632,24 +633,34 @@ public class SqlJsonUtils {
return jsonValueExpression(input);
}
- /** Parses a JSON string into a {@link Variant}, throwing on malformed
input. */
- public static Variant parseJsonToVariant(StringData jsonStr, boolean
allowDuplicateKeys) {
+ /**
+ * Parses a JSON string into a {@link Variant}.
+ *
+ * <p>Malformed input fails with the parser's own exception, which {@code
TRY_PARSE_JSON} can
+ * drop cheaply. {@code PARSE_JSON} reports it through {@link
#parseJsonError}.
+ */
+ public static Variant parseJsonToVariant(StringData jsonStr, boolean
allowDuplicateKeys)
+ throws IOException {
+ return BinaryVariantInternalBuilder.parseJson(jsonStr.toBytes(),
allowDuplicateKeys);
+ }
+
+ /**
+ * Builds the error that {@code PARSE_JSON} throws when {@link
#parseJsonToVariant} rejects
+ * {@code jsonStr}. Input that is not valid UTF-8 is reported by the index
of its first invalid
+ * byte rather than by its content.
+ */
+ public static TableRuntimeException parseJsonError(StringData jsonStr,
Throwable cause) {
final byte[] bytes = jsonStr.toBytes();
- try {
- return BinaryVariantInternalBuilder.parseJson(bytes,
allowDuplicateKeys);
- } catch (Exception e) {
- final int invalidIndex =
- StringUtf8Utils.firstInvalidUtf8ByteIndex(bytes, 0,
bytes.length);
- if (invalidIndex >= 0) {
- throw new TableRuntimeException(
- String.format(
- "Failed to parse JSON string: Invalid UTF-8
byte at index %d of %d.",
- invalidIndex, bytes.length),
- e);
- }
- throw new TableRuntimeException(
- String.format("Failed to parse JSON string: %s", jsonStr),
e);
+ final int invalidIndex =
StringUtf8Utils.firstInvalidUtf8ByteIndex(bytes, 0, bytes.length);
+ if (invalidIndex >= 0) {
+ return new TableRuntimeException(
+ String.format(
+ "Failed to parse JSON string: Invalid UTF-8 byte
at index %d of %d.",
+ invalidIndex, bytes.length),
+ cause);
}
+ return new TableRuntimeException(
+ String.format("Failed to parse JSON string: %s", jsonStr),
cause);
}
private static JsonValueContext jsonValueExpression(String input) {
