This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 13b74a36de2 [FLINK-40996][table] Make `TRY_PARSE_JSON`, `PARSE_JSON`
reuse parsed json
13b74a36de2 is described below
commit 13b74a36de2a88aa39ff23b482995b94e55e2c92
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Fri Oct 16 17:14:46 2026 +0200
[FLINK-40996][table] Make `TRY_PARSE_JSON`, `PARSE_JSON` reuse parsed json
---
.../functions/BuiltInFunctionDefinitions.java | 6 +-
.../table/planner/codegen/JsonCodeGenUtils.java | 208 +++++++++++++++++++++
.../table/planner/codegen/ExprCodeGenerator.scala | 6 +
.../planner/codegen/calls/BuiltInMethods.scala | 8 +-
.../table/planner/codegen/JsonParseReuseTest.java | 102 +++++++++-
.../table/runtime/functions/SqlJsonUtils.java | 23 +++
.../functions/scalar/ParseJsonFunction.java | 67 -------
.../functions/scalar/TryParseJsonFunction.java | 54 ------
8 files changed, 343 insertions(+), 131 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 791d19b2fdb..9e993e26429 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -3215,8 +3215,7 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(
nullableIfArgs(
ConstantArgumentCount.of(0),
explicit(DataTypes.VARIANT())))
- .runtimeClass(
-
"org.apache.flink.table.runtime.functions.scalar.ParseJsonFunction")
+ .runtimeProvided()
.build();
public static final BuiltInFunctionDefinition TRY_PARSE_JSON =
@@ -3230,8 +3229,7 @@ public final class BuiltInFunctionDefinitions {
logical(LogicalTypeFamily.CHARACTER_STRING),
logical(LogicalTypeRoot.BOOLEAN))))
.outputTypeStrategy(forceNullable(explicit(DataTypes.VARIANT())))
- .runtimeClass(
-
"org.apache.flink.table.runtime.functions.scalar.TryParseJsonFunction")
+ .runtimeProvided()
.build();
public static final BuiltInFunctionDefinition PARSE_XML =
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 f63c35085ea..2dd2f068c71 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
@@ -22,16 +22,224 @@ import
org.apache.flink.table.planner.codegen.calls.BuiltInMethods;
import org.apache.flink.table.planner.codegen.calls.JsonParseReuse;
import org.apache.flink.table.runtime.functions.SqlJsonUtils;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.types.variant.Variant;
import java.lang.reflect.Method;
+import scala.Option;
import scala.collection.Seq;
/** Utilities for the code generation of JSON functions. */
public final class JsonCodeGenUtils {
+ private static final int PARSED_VALUE_INDEX = Integer.MIN_VALUE;
+ private static final int PARSED_ERROR_INDEX = Integer.MIN_VALUE + 1;
+
private JsonCodeGenUtils() {}
+ /**
+ * Generates {@code PARSE_JSON(str[, allowDuplicateKeys])} ({@code isTry
== false}) or {@code
+ * TRY_PARSE_JSON(...)} ({@code isTry == true}).
+ *
+ * <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.
+ *
+ * <p>For {@code PARSE_JSON(str)} the generated code looks like:
+ *
+ * <pre>{@code
+ * // reusable members shared by every call on the same (input, allowDup):
+ * Variant variantParsed$0;
+ * Throwable variantParseError$0;
+ * StringData variantParseInput$0;
+ * java.lang.Boolean variantParseDup$0;
+ *
+ * private Variant parseVariant$0(StringData in, java.lang.Boolean
allowDup) {
+ * if (in == variantParseInput$0 && java.util.Objects.equals(allowDup,
variantParseDup$0)) {
+ * return variantParsed$0;
+ * }
+ * variantParseInput$0 = in;
+ * variantParseDup$0 = allowDup;
+ * variantParseError$0 = null;
+ * try {
+ * variantParsed$0 =
+ * SqlJsonUtils.parseJsonToVariant(in,
java.lang.Boolean.TRUE.equals(allowDup));
+ * } catch (Exception e) {
+ * variantParsed$0 = null;
+ * variantParseError$0 = t;
+ * }
+ * return variantParsed$0;
+ * }
+ *
+ * // per call site:
+ * result$0 = null;
+ * isNull$0 = true;
+ * 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);
+ * }
+ * isNull$0 = result$0 == null;
+ * }
+ * }</pre>
+ */
+ public static GeneratedExpression generateParseJson(
+ CodeGeneratorContext ctx,
+ LogicalType returnType,
+ Seq<GeneratedExpression> operands,
+ boolean isTry) {
+ final GeneratedExpression input = operands.apply(0);
+ final GeneratedExpression allowDup = operands.length() > 1 ?
operands.apply(1) : null;
+
+ final String key =
+ "variantParse:"
+ + input.resultTerm()
+ + ":"
+ + (allowDup == null ? "" : allowDup.resultTerm());
+
+ final Option<GeneratedExpression> cachedValue =
+ ctx.getReusableInputUnboxingExprs(key, PARSED_VALUE_INDEX);
+ final Option<GeneratedExpression> cachedError =
+ ctx.getReusableInputUnboxingExprs(key, PARSED_ERROR_INDEX);
+
+ final String valueTerm;
+ final String errorTerm;
+ if (cachedValue.isDefined() && cachedError.isDefined()) {
+ valueTerm = cachedValue.get().resultTerm();
+ errorTerm = cachedError.get().resultTerm();
+ } else {
+ final String variantType = CodeGenUtils.className(Variant.class);
+ final String inputType =
CodeGenUtils.boxedTypeTermForType(input.resultType());
+ final String parsed = CodeGenUtils.newName(ctx, "variantParsed");
+ final String error = CodeGenUtils.newName(ctx,
"variantParseError");
+ final String lastInput = CodeGenUtils.newName(ctx,
"variantParseInput");
+ final String lastDup = CodeGenUtils.newName(ctx,
"variantParseDup");
+ final String method = CodeGenUtils.newName(ctx, "parseVariant");
+
+ ctx.addReusableMember(variantType + " " + parsed + ";");
+ ctx.addReusableMember("Throwable " + error + ";");
+ ctx.addReusableMember(inputType + " " + lastInput + ";");
+ ctx.addReusableMember("java.lang.Boolean " + lastDup + ";");
+ ctx.addReusableMember(
+ "private "
+ + variantType
+ + " "
+ + method
+ + "("
+ + inputType
+ + " in, java.lang.Boolean allowDup) {\n"
+ + " if (in == "
+ + lastInput
+ + " && java.util.Objects.equals(allowDup, "
+ + lastDup
+ + ")) {\n"
+ + " return "
+ + parsed
+ + ";\n"
+ + " }\n"
+ + " "
+ + lastInput
+ + " = in;\n"
+ + " "
+ + lastDup
+ + " = allowDup;\n"
+ + " "
+ + error
+ + " = null;\n"
+ + " try {\n"
+ + " "
+ + parsed
+ + " = "
+ +
CodeGenUtils.qualifyMethod(BuiltInMethods.PARSE_JSON())
+ + "(in,
java.lang.Boolean.TRUE.equals(allowDup));\n"
+ + " } catch (Exception e) {\n"
+ + " "
+ + parsed
+ + " = null;\n"
+ + " "
+ + error
+ + " = e;\n"
+ + " }\n"
+ + " return "
+ + parsed
+ + ";\n"
+ + "}\n");
+
+ final String dupArg =
+ allowDup == null
+ ? "null"
+ : "("
+ + allowDup.nullTerm()
+ + ") ? (java.lang.Boolean) null :
java.lang.Boolean.valueOf("
+ + allowDup.resultTerm()
+ + ")";
+ valueTerm = method + "(" + input.resultTerm() + ", " + dupArg +
")";
+ errorTerm = error;
+
+ ctx.addReusableInputUnboxingExprs(
+ key,
+ PARSED_VALUE_INDEX,
+ new GeneratedExpression(
+ valueTerm,
+ GeneratedExpression.NEVER_NULL(),
+ GeneratedExpression.NO_CODE(),
+ null,
+ Option.<Object>empty()));
+ ctx.addReusableInputUnboxingExprs(
+ key,
+ PARSED_ERROR_INDEX,
+ new GeneratedExpression(
+ error,
+ GeneratedExpression.NEVER_NULL(),
+ GeneratedExpression.NO_CODE(),
+ null,
+ Option.<Object>empty()));
+ }
+
+ final String resultTerm =
+ ctx.addReusableLocalVariable(
+ CodeGenUtils.primitiveTypeTermForType(returnType),
"result");
+ final String nullTerm = ctx.addReusableLocalVariable("boolean",
"isNull");
+
+ final String rethrow =
+ isTry
+ ? ""
+ : " if ("
+ + errorTerm
+ + " != null) {\n"
+ + "
org.apache.flink.util.ExceptionUtils.rethrow("
+ + errorTerm
+ + ");\n"
+ + " }\n";
+
+ final String code =
+ input.code()
+ + "\n"
+ + (allowDup == null ? "" : allowDup.code() + "\n")
+ + resultTerm
+ + " = null;\n"
+ + nullTerm
+ + " = true;\n"
+ + "if (!"
+ + input.nullTerm()
+ + ") {\n"
+ + " "
+ + resultTerm
+ + " = "
+ + valueTerm
+ + ";\n"
+ + rethrow
+ + " "
+ + nullTerm
+ + " = "
+ + resultTerm
+ + " == null;\n"
+ + "}\n";
+
+ return new GeneratedExpression(
+ resultTerm, nullTerm, code, returnType,
Option.<Object>empty());
+ }
+
/**
* Generates {@code JSON_TYPE(jsonValue)} or {@code JSON_TYPE(jsonValue,
path)}.
*
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
index 1053e21741e..8b66910d2aa 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
@@ -948,6 +948,12 @@ class ExprCodeGenerator(
case BuiltInFunctionDefinitions.JSON_STRING =>
new JsonStringCallGen(call, rexProgram).generate(ctx, operands,
resultType)
+ case BuiltInFunctionDefinitions.PARSE_JSON =>
+ JsonCodeGenUtils.generateParseJson(ctx, resultType, operands,
false)
+
+ case BuiltInFunctionDefinitions.TRY_PARSE_JSON =>
+ JsonCodeGenUtils.generateParseJson(ctx, resultType, operands, true)
+
case BuiltInFunctionDefinitions.JSON_LENGTH =>
JsonCodeGenUtils.generateJsonLength(ctx, resultType, operands)
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 d9378d7116f..9bb1de380a5 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
@@ -18,7 +18,7 @@
package org.apache.flink.table.planner.codegen.calls
import org.apache.flink.table.api.{JsonExistsOnError, JsonQueryOnEmptyOrError,
JsonQueryWrapper, JsonValueOnEmptyOrError}
-import org.apache.flink.table.data.{DecimalData, DecimalDataUtils,
TimestampData}
+import org.apache.flink.table.data.{DecimalData, DecimalDataUtils, StringData,
TimestampData}
import org.apache.flink.table.data.binary.{BinaryStringData,
BinaryStringDataUtil}
import org.apache.flink.table.functions.SqlLikeUtils
import org.apache.flink.table.runtime.functions._
@@ -514,6 +514,12 @@ object BuiltInMethods {
val JSON_PARSE =
Types.lookupMethod(classOf[SqlJsonUtils], "jsonParse", classOf[String])
+ val PARSE_JSON = Types.lookupMethod(
+ classOf[SqlJsonUtils],
+ "parseJsonToVariant",
+ classOf[StringData],
+ classOf[Boolean])
+
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 cf7e19fd4f2..a429e541911 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
@@ -37,10 +37,12 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests that multiple JSON function calls on the same input reuse the parsed
JSON. */
class JsonParseReuseTest {
@@ -48,6 +50,9 @@ class JsonParseReuseTest {
private static final Pattern JSON_PARSE_PATTERN =
Pattern.compile("\\b" +
Pattern.quote(BuiltInMethods.JSON_PARSE().getName()) + "\\(");
+ private static final Pattern PARSE_JSON_PATTERN =
+ Pattern.compile("\\b" +
Pattern.quote(BuiltInMethods.PARSE_JSON().getName()) + "\\(");
+
private StreamTableEnvironment tEnv;
private static final String JSON_ROW1 =
@@ -72,7 +77,20 @@ class JsonParseReuseTest {
}
private static int countJsonParse(final String code) {
- return GeneratedCodeTestUtils.countMatches(JSON_PARSE_PATTERN, code);
+ return count(JSON_PARSE_PATTERN, code);
+ }
+
+ private static int countParseJson(final String code) {
+ return count(PARSE_JSON_PATTERN, code);
+ }
+
+ private static int count(final Pattern pattern, final String code) {
+ final Matcher m = pattern.matcher(code);
+ int count = 0;
+ while (m.find()) {
+ count++;
+ }
+ return count;
}
private List<String> generatedClassCodes(final String sql) {
@@ -185,15 +203,25 @@ class JsonParseReuseTest {
"SELECT JSON_VALUE(TRIM(json_data), '$.type'), "
+ "JSON_QUERY(TRIM(json_data), '$.address'), "
+ "JSON_TYPE(TRIM(json_data)), "
- + "JSON_LENGTH(TRIM(json_data)) FROM json_src";
+ + "JSON_LENGTH(TRIM(json_data)), "
+ + "JSON_STRING(PARSE_JSON(TRIM(json_data))), "
+ + "JSON_STRING(TRY_PARSE_JSON(TRIM(json_data))) FROM
json_src";
final List<Row> rows = collect(sql);
+ // PARSE_JSON returns a VARIANT, which sorts object keys
+ final String variant1 =
+
"{\"address\":{\"city\":\"Munich\"},\"age\":42,\"roles\":[\"user\",\"viewer\"],\"type\":\"account\"}";
+ final String variant2 =
+
"{\"address\":{\"city\":\"Berlin\"},\"age\":30,\"roles\":[\"admin\"],\"type\":\"admin\"}";
assertThat(rows)
.containsExactlyInAnyOrder(
- Row.of("account", "{\"city\":\"Munich\"}", "object",
4),
- Row.of("admin", "{\"city\":\"Berlin\"}", "object", 4));
+ Row.of("account", "{\"city\":\"Munich\"}", "object",
4, variant1, variant1),
+ Row.of("admin", "{\"city\":\"Berlin\"}", "object", 4,
variant2, variant2));
final String code = extractGeneratedCode(sql);
assertThat(countJsonParse(code))
- .as("Calls on the same computed input should parse once")
+ .as("JSON_VALUE-family calls on the same computed input should
parse once")
+ .isOne();
+ assertThat(countParseJson(code))
+ .as("PARSE_JSON + TRY_PARSE_JSON on the same computed input
should parse once")
.isOne();
}
@@ -482,4 +510,68 @@ class JsonParseReuseTest {
.as("JSON_TYPE + JSON_VALUE on the same input should parse
once")
.isOne();
}
+
+ @Test
+ void testParseJsonAndTryParseJsonShareParse() {
+ final String sql =
+ "SELECT JSON_STRING(PARSE_JSON(json_data)), "
+ + "JSON_STRING(TRY_PARSE_JSON(json_data)) FROM
json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).hasSize(2).allSatisfy(row ->
assertThat(row.getField(0)).isNotNull());
+ assertThat(rows).allSatisfy(row ->
assertThat(row.getField(0)).isEqualTo(row.getField(1)));
+ assertThat(countParseJson(extractGeneratedCode(sql)))
+ .as("PARSE_JSON + TRY_PARSE_JSON on the same input should
parse once")
+ .isOne();
+ }
+
+ @Test
+ void testParseErrorRethrowForParseJson() {
+ 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)))
+ .as("PARSE_JSON + TRY_PARSE_JSON on the same input should
parse once")
+ .isOne();
+ assertThatThrownBy(() -> collect(sharedSql))
+ .hasStackTraceContaining("Failed to parse JSON string");
+
+ assertThat(collect("SELECT JSON_STRING(TRY_PARSE_JSON(j)) FROM " +
badSrc))
+ .containsExactly(Row.of((String) null));
+ }
+
+ @Test
+ void testParseJsonDifferentInputsParseSeparately() {
+ final String sql =
+ "SELECT JSON_STRING(PARSE_JSON(json_data)), "
+ + "JSON_STRING(TRY_PARSE_JSON(other_json)) FROM
json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).hasSize(2);
+ assertThat(countParseJson(extractGeneratedCode(sql)))
+ .as("PARSE_JSON calls on different inputs should parse
separately")
+ .isEqualTo(2);
+ }
+
+ @Test
+ void testParseJsonSameAllowDuplicateKeysSharesParse() {
+ final String sql =
+ "SELECT JSON_STRING(PARSE_JSON(json_data, true)), "
+ + "JSON_STRING(TRY_PARSE_JSON(json_data, true)) FROM
json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).hasSize(2);
+ assertThat(countParseJson(extractGeneratedCode(sql)))
+ .as("same input and allowDuplicateKeys should parse once")
+ .isOne();
+ }
+
+ @Test
+ void testParseJsonDifferentAllowDuplicateKeysParseSeparately() {
+ final String sql =
+ "SELECT JSON_STRING(PARSE_JSON(json_data, true)), "
+ + "JSON_STRING(TRY_PARSE_JSON(json_data, false)) FROM
json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).hasSize(2);
+ assertThat(countParseJson(extractGeneratedCode(sql)))
+ .as("different allowDuplicateKeys should parse separately")
+ .isEqualTo(2);
+ }
}
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 bcbf7a3f5f2..b524854b574 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
@@ -26,6 +26,9 @@ import org.apache.flink.table.api.JsonValueOnEmptyOrError;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.StringUtf8Utils;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.types.variant.Variant;
import org.apache.flink.shaded.com.jayway.jsonpath.Configuration;
import org.apache.flink.shaded.com.jayway.jsonpath.DocumentContext;
@@ -611,6 +614,26 @@ 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) {
+ 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);
+ }
+ }
+
private static JsonValueContext jsonValueExpression(String input) {
try {
return JsonValueContext.withJavaObj(dejsonize(input));
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
deleted file mode 100644
index 7c2f5789c8f..00000000000
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.scalar;
-
-import org.apache.flink.table.api.TableRuntimeException;
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.binary.StringUtf8Utils;
-import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
-import org.apache.flink.table.functions.SpecializedFunction;
-import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
-import org.apache.flink.types.variant.Variant;
-
-import javax.annotation.Nullable;
-
-/** Implementation of {@link BuiltInFunctionDefinitions#PARSE_JSON}. */
-public class ParseJsonFunction extends BuiltInScalarFunction {
-
- public ParseJsonFunction(SpecializedFunction.SpecializedContext context) {
- super(BuiltInFunctionDefinitions.PARSE_JSON, context);
- }
-
- public @Nullable Variant eval(@Nullable StringData jsonStr) {
- return eval(jsonStr, false);
- }
-
- /** A {@code NULL} {@code allowDuplicateKeys} is treated as {@code false}.
*/
- public @Nullable Variant eval(
- @Nullable StringData jsonStr, @Nullable Boolean
allowDuplicateKeys) {
- if (jsonStr == null) {
- return null;
- }
-
- final byte[] bytes = jsonStr.toBytes();
- try {
- return BinaryVariantInternalBuilder.parseJson(
- bytes, Boolean.TRUE.equals(allowDuplicateKeys));
- } catch (Throwable 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);
- }
- }
-}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
deleted file mode 100644
index c3e4dc66bef..00000000000
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.scalar;
-
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
-import org.apache.flink.table.functions.SpecializedFunction;
-import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
-import org.apache.flink.types.variant.Variant;
-
-import javax.annotation.Nullable;
-
-/** Implementation of {@link BuiltInFunctionDefinitions#TRY_PARSE_JSON}. */
-public class TryParseJsonFunction extends BuiltInScalarFunction {
-
- public TryParseJsonFunction(SpecializedFunction.SpecializedContext
context) {
- super(BuiltInFunctionDefinitions.TRY_PARSE_JSON, context);
- }
-
- public @Nullable Variant eval(@Nullable StringData jsonStr) {
- return eval(jsonStr, false);
- }
-
- /** A {@code NULL} {@code allowDuplicateKeys} is treated as {@code false}.
*/
- public @Nullable Variant eval(
- @Nullable StringData jsonStr, @Nullable Boolean
allowDuplicateKeys) {
- if (jsonStr == null) {
- return null;
- }
-
- try {
- return BinaryVariantInternalBuilder.parseJson(
- jsonStr.toBytes(),
Boolean.TRUE.equals(allowDuplicateKeys));
- } catch (Throwable e) {
- return null;
- }
- }
-}
