You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/10/01 14:04:08 UTC
[flink] 02/02: [FLINK-16204][table] Support JSON_ARRAY()
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f612a66c06b733eeba9e3c37e8760f329a7ac742
Author: Ingo Bürk <in...@tngtech.com>
AuthorDate: Fri Sep 17 12:11:24 2021 +0200
[FLINK-16204][table] Support JSON_ARRAY()
This closes #17313.
---
docs/data/sql_functions.yml | 33 +++++
docs/data/sql_functions_zh.yml | 32 +++++
flink-python/pyflink/table/expressions.py | 38 +++++-
.../org/apache/flink/table/api/Expressions.java | 45 +++++++
.../table/api/ImplicitExpressionConversions.scala | 47 ++++++-
.../org/apache/flink/table/api/JsonOnNull.java | 5 +-
.../functions/BuiltInFunctionDefinitions.java | 9 ++
.../strategies/SpecificInputTypeStrategies.java | 13 ++
.../converter/converters/CustomizedConverters.java | 1 +
...bjectConverter.java => JsonArrayConverter.java} | 37 +-----
.../converter/converters/JsonConverterUtil.java | 52 ++++++++
.../converter/converters/JsonObjectConverter.java | 18 +--
.../functions/sql/FlinkSqlOperatorTable.java | 1 +
.../table/planner/codegen/ExprCodeGenerator.scala | 2 +
.../table/planner/codegen/ExpressionReducer.scala | 8 +-
.../table/planner/codegen/JsonGenerateUtils.scala | 33 ++++-
.../planner/codegen/calls/BuiltInMethods.scala | 3 -
.../planner/codegen/calls/JsonArrayCallGen.scala | 85 +++++++++++++
.../planner/codegen/calls/JsonObjectCallGen.scala | 28 ++---
.../planner/functions/JsonFunctionsITCase.java | 139 ++++++++++++++++++++-
.../table/runtime/functions/SqlJsonUtils.java | 6 +
21 files changed, 547 insertions(+), 88 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index b3ddb15..438aeab 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -739,6 +739,10 @@ json:
This function returns a JSON string. The `ON NULL` behavior defines how to treat `NULL`
values. If omitted, `NULL ON NULL` is assumed by default.
+ Values which are created from another JSON construction function call (`JSON_OBJECT`,
+ `JSON_ARRAY`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
+
```
-- '{}'
JSON_OBJECT()
@@ -762,6 +766,35 @@ json:
)
)
```
+ - sql: JSON_ARRAY([value]* [ { NULL | ABSENT } ON NULL ])
+ table: jsonArray(JsonOnNull, values...)
+ description: |
+ Builds a JSON array string from a list of values.
+
+ This function returns a JSON string. The values can be arbitrary expressions. The `ON NULL`
+ behavior defines how to treat `NULL` values. If omitted, `ABSENT ON NULL` is assumed by
+ default.
+
+ Elements which are created from another JSON construction function call (`JSON_OBJECT`,
+ `JSON_ARRAY`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
+
+ ```
+ -- '[]'
+ JSON_ARRAY()
+ -- '[1,"2"]'
+ JSON_ARRAY(1, '2')
+
+ -- Expressions as values
+ JSON_ARRAY(orders.orderId)
+
+ -- ON NULL
+ JSON_ARRAY(CAST(NULL AS STRING) NULL ON NULL) -- '[null]'
+ JSON_ARRAY(CAST(NULL AS STRING) ABSENT ON NULL) -- '[]'
+
+ -- '[[1]]'
+ JSON_ARRAY(JSON_ARRAY(1))
+ ```
valueconstruction:
- sql: |
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index f0fa6b4..ade20f4 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -857,6 +857,9 @@ json:
This function returns a JSON string. The `ON NULL` behavior defines how to treat `NULL`
values. If omitted, `NULL ON NULL` is assumed by default.
+ Values which are created from another JSON construction function call (`JSON_OBJECT`,
+ `JSON_ARRAY`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
```
-- '{}'
@@ -881,6 +884,35 @@ json:
)
)
```
+ - sql: JSON_ARRAY([value]* [ { NULL | ABSENT } ON NULL ])
+ table: jsonArray(JsonOnNull, values...)
+ description: |
+ Builds a JSON array string from a list of values.
+
+ This function returns a JSON string. The values can be arbitrary expressions. The `ON NULL`
+ behavior defines how to treat `NULL` values. If omitted, `ABSENT ON NULL` is assumed by
+ default.
+
+ Elements which are created from another JSON construction function call (`JSON_OBJECT`,
+ `JSON_ARRAY`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
+
+ ```
+ -- '[]'
+ JSON_ARRAY()
+ -- '[1,"2"]'
+ JSON_ARRAY(1, '2')
+
+ -- Expressions as values
+ JSON_ARRAY(orders.orderId)
+
+ -- ON NULL
+ JSON_ARRAY(CAST(NULL AS STRING) NULL ON NULL) -- '[null]'
+ JSON_ARRAY(CAST(NULL AS STRING) ABSENT ON NULL) -- '[]'
+
+ -- '[[1]]'
+ JSON_ARRAY(JSON_ARRAY(1))
+ ```
valueconstruction:
- sql: |
diff --git a/flink-python/pyflink/table/expressions.py b/flink-python/pyflink/table/expressions.py
index 56ff899..55bb889 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -30,7 +30,7 @@ __all__ = ['if_then_else', 'lit', 'col', 'range_', 'and_', 'or_', 'not_', 'UNBOU
'temporal_overlaps', 'date_format', 'timestamp_diff', 'array', 'row', 'map_',
'row_interval', 'pi', 'e', 'rand', 'rand_integer', 'atan2', 'negative', 'concat',
'concat_ws', 'uuid', 'null_of', 'log', 'with_columns', 'without_columns', 'json_object',
- 'call', 'call_sql', 'source_watermark']
+ 'json_array', 'call', 'call_sql', 'source_watermark']
def _leaf_op(op_name: str) -> Expression:
@@ -619,6 +619,10 @@ def json_object(on_null: JsonOnNull = JsonOnNull.NULL, *args) -> Expression:
This function returns a JSON string. The `on_null` behavior defines how to treat `NULL` values.
+ Values which are created from another JSON construction function call (`json_object`,
+ `json_array`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
+
Examples:
::
@@ -633,10 +637,42 @@ def json_object(on_null: JsonOnNull = JsonOnNull.NULL, *args) -> Expression:
>>> # '{"K1":{"K2":"V"}}'
>>> json_object(JsonOnNull.NULL, "K1", json_object(JsonOnNull.NULL, "K2", "V"))
+
+ .. seealso:: :func:`~pyflink.table.expressions.json_array`
"""
return _varargs_op("jsonObject", *(on_null._to_j_json_on_null(), *args))
+def json_array(on_null: JsonOnNull = JsonOnNull.ABSENT, *args) -> Expression:
+ """
+ Builds a JSON array string from a list of values.
+
+ This function returns a JSON string. The values can be arbitrary expressions. The `on_null`
+ behavior defines how to treat `NULL` values.
+
+ Elements which are created from another JSON construction function call (`json_object`,
+ `json_array`) are inserted directly rather than as a string. This allows building nested JSON
+ structures.
+
+ Examples:
+ ::
+
+ >>> json_array() # '[]'
+ >>> json_array(JsonOnNull.NULL, 1, "2") # '[1,"2"]'
+
+ >>> # Expressions as values
+ >>> json_array(JsonOnNull.NULL, col("orderId"))
+
+ >>> json_array(JsonOnNull.NULL, null_of(DataTypes.STRING())) # '[null]'
+ >>> json_array(JsonOnNull.ABSENT, null_of(DataTypes.STRING())) # '[]'
+
+ >>> json_array(JsonOnNull.NULL, json_array(JsonOnNull.NULL, 1)) # '[[1]]'
+
+ .. seealso:: :func:`~pyflink.table.expressions.json_object`
+ """
+ return _varargs_op("jsonArray", *(on_null._to_j_json_on_null(), *args))
+
+
def call(f: Union[str, UserDefinedFunctionWrapper], *args) -> Expression:
"""
The first parameter `f` could be a str or a Python user-defined function.
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index c73046a..8958f26 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -43,6 +43,7 @@ import static org.apache.flink.table.expressions.ApiExpressionUtils.objectToExpr
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_ARRAY;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_OBJECT;
/**
@@ -594,6 +595,10 @@ public final class Expressions {
* <p>This function returns a JSON string. The {@link JsonOnNull onNull} behavior defines how to
* treat {@code NULL} values.
*
+ * <p>Values which are created from another JSON construction function call ({@code jsonObject},
+ * {@code jsonArray}) are inserted directly rather than as a string. This allows building nested
+ * JSON structures.
+ *
* <p>Examples:
*
* <pre>{@code
@@ -613,6 +618,8 @@ public final class Expressions {
* // {"K1":{"K2":"V"}}
* jsonObject(JsonOnNull.NULL, "K1", jsonObject(JsonOnNull.NULL, "K2", "V"))
* }</pre>
+ *
+ * @see #jsonArray(JsonOnNull, Object...)
*/
public static ApiExpression jsonObject(JsonOnNull onNull, Object... keyValues) {
final Object[] arguments =
@@ -622,6 +629,44 @@ public final class Expressions {
}
/**
+ * Builds a JSON array string from a list of values.
+ *
+ * <p>This function returns a JSON string. The values can be arbitrary expressions. The {@link
+ * JsonOnNull onNull} behavior defines how to treat {@code NULL} values.
+ *
+ * <p>Elements which are created from another JSON construction function call ({@code
+ * jsonObject}, {@code jsonArray}) are inserted directly rather than as a string. This allows
+ * building nested JSON structures.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // "[]"
+ * jsonArray(JsonOnNull.NULL)
+ * // "[1,\"2\"]"
+ * jsonArray(JsonOnNull.NULL, 1, "2")
+ *
+ * // Expressions as values
+ * jsonArray(JsonOnNull.NULL, $("orderId"))
+ *
+ * // ON NULL
+ * jsonArray(JsonOnNull.NULL, nullOf(DataTypes.STRING())) // "[null]"
+ * jsonArray(JsonOnNull.ABSENT, nullOf(DataTypes.STRING())) // "[]"
+ *
+ * // "[[1]]"
+ * jsonArray(JsonOnNull.NULL, jsonArray(JsonOnNull.NULL, 1))
+ * }</pre>
+ *
+ * @see #jsonObject(JsonOnNull, Object...)
+ */
+ public static ApiExpression jsonArray(JsonOnNull onNull, Object... values) {
+ final Object[] arguments =
+ Stream.concat(Stream.of(onNull), Arrays.stream(values)).toArray(Object[]::new);
+
+ return apiCall(JSON_ARRAY, arguments);
+ }
+
+ /**
* A call to a function that will be looked up in a catalog. There are two kinds of functions:
*
* <ul>
diff --git a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
index 316065b..19fee24 100644
--- a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
+++ b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
@@ -810,11 +810,15 @@ trait ImplicitExpressionConversions {
/**
* Builds a JSON object string from a list of key-value pairs.
*
- * <code>keyValues</code> is an even-numbered list of alternating key/value pairs. Note that keys
- * must be string literals, values may be arbitrary expressions.
+ * `keyValues` is an even-numbered list of alternating key/value pairs. Note that keys must be
+ * string literals, values may be arbitrary expressions.
*
* This function returns a JSON string. The [[JsonOnNull onNull]] behavior defines how to treat
- * <code>NULL</code> values.
+ * `NULL` values.
+ *
+ * Values which are created from another JSON construction function call
+ * (`jsonObject`, `jsonArray`) are inserted directly rather than as a string. This allows
+ * building nested JSON structures.
*
* Examples:
* {{{
@@ -833,8 +837,45 @@ trait ImplicitExpressionConversions {
* // {"K1":{"K2":"V"}}
* jsonObject(JsonOnNull.NULL, "K1", jsonObject(JsonOnNull.NULL, "K2", "V"))
* }}}
+ *
+ * @see #jsonObject
*/
def jsonObject(onNull: JsonOnNull, keyValues: Expression*): Expression = {
Expressions.jsonObject(onNull, keyValues)
}
+
+ /**
+ * Builds a JSON array string from a list of values.
+ *
+ * This function returns a JSON string. The values can be arbitrary expressions. The
+ * [[JsonOnNull onNull]] behavior defines how to treat `NULL` values.
+ *
+ * Elements which are created from another JSON construction function call
+ * (`jsonObject`, `jsonArray`) are inserted directly rather than as a string. This allows
+ * building nested JSON structures.
+ *
+ * Examples:
+ *
+ * {{{
+ * // "[]"
+ * jsonArray(JsonOnNull.NULL)
+ * // "[1,\"2\"]"
+ * jsonArray(JsonOnNull.NULL, 1, "2")
+ *
+ * // Expressions as values
+ * jsonArray(JsonOnNull.NULL, $("orderId"))
+ *
+ * // ON NULL
+ * jsonArray(JsonOnNull.NULL, nullOf(DataTypes.STRING())) // "[null]"
+ * jsonArray(JsonOnNull.ABSENT, nullOf(DataTypes.STRING())) // "[]"
+ *
+ * // "[[1]]"
+ * jsonArray(JsonOnNull.NULL, jsonArray(JsonOnNull.NULL, 1))
+ * }}}
+ *
+ * @see #jsonObject
+ */
+ def jsonArray(onNull: JsonOnNull, values: Expression*): Expression = {
+ Expressions.jsonArray(onNull, values)
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/JsonOnNull.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/JsonOnNull.java
index 78c98bc..076303d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/JsonOnNull.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/JsonOnNull.java
@@ -22,7 +22,10 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.expressions.TableSymbol;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
-/** Behavior for entries with a null value for {@link BuiltInFunctionDefinitions#JSON_OBJECT}. */
+/**
+ * Behavior for entries with a null value for {@link BuiltInFunctionDefinitions#JSON_OBJECT} and
+ * {@link BuiltInFunctionDefinitions#JSON_ARRAY}.
+ */
@PublicEvolving
public enum JsonOnNull implements TableSymbol {
/** Use a JSON {@code null} value. */
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 a0e257f..da41c5e 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
@@ -1578,6 +1578,15 @@ public final class BuiltInFunctionDefinitions {
.runtimeDeferred()
.build();
+ public static final BuiltInFunctionDefinition JSON_ARRAY =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("JSON_ARRAY")
+ .kind(SCALAR)
+ .inputTypeStrategy(SpecificInputTypeStrategies.JSON_ARRAY)
+ .outputTypeStrategy(explicit(DataTypes.STRING().notNull()))
+ .runtimeDeferred()
+ .build();
+
// --------------------------------------------------------------------------------------------
// Other functions
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index e9d62db..3e0cc3c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -36,6 +36,7 @@ import static org.apache.flink.table.types.inference.InputTypeStrategies.logical
import static org.apache.flink.table.types.inference.InputTypeStrategies.or;
import static org.apache.flink.table.types.inference.InputTypeStrategies.repeatingSequence;
import static org.apache.flink.table.types.inference.InputTypeStrategies.symbol;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.varyingSequence;
/**
* Entry point for specific input type strategies not covered in {@link InputTypeStrategies}.
@@ -76,6 +77,18 @@ public final class SpecificInputTypeStrategies {
logical(LogicalTypeRoot.BOOLEAN),
logical(LogicalTypeFamily.NUMERIC))));
+ /** Input strategy for {@link BuiltInFunctionDefinitions#JSON_ARRAY}. */
+ public static final InputTypeStrategy JSON_ARRAY =
+ varyingSequence(
+ symbol(JsonOnNull.class),
+ or(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.BINARY_STRING),
+ logical(LogicalTypeFamily.TIMESTAMP),
+ logical(LogicalTypeFamily.CONSTRUCTED),
+ logical(LogicalTypeRoot.BOOLEAN),
+ logical(LogicalTypeFamily.NUMERIC)));
+
// --------------------------------------------------------------------------------------------
// Strategies composed of other strategies
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/CustomizedConverters.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/CustomizedConverters.java
index 390623d..fa90d6d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/CustomizedConverters.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/CustomizedConverters.java
@@ -57,6 +57,7 @@ public class CustomizedConverters {
CONVERTERS.put(BuiltInFunctionDefinitions.JSON_VALUE, new JsonValueConverter());
CONVERTERS.put(BuiltInFunctionDefinitions.JSON_QUERY, new JsonQueryConverter());
CONVERTERS.put(BuiltInFunctionDefinitions.JSON_OBJECT, new JsonObjectConverter());
+ CONVERTERS.put(BuiltInFunctionDefinitions.JSON_ARRAY, new JsonArrayConverter());
CONVERTERS.put(InternalFunctionDefinitions.THROW_EXCEPTION, new ThrowExceptionConverter());
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonArrayConverter.java
similarity index 53%
copy from flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java
copy to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonArrayConverter.java
index da1c378..b0eee09 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonArrayConverter.java
@@ -19,11 +19,7 @@
package org.apache.flink.table.planner.expressions.converter.converters;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.JsonOnNull;
-import org.apache.flink.table.api.TableException;
import org.apache.flink.table.expressions.CallExpression;
-import org.apache.flink.table.expressions.Expression;
-import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.planner.expressions.converter.CallExpressionConvertRule;
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable;
@@ -34,44 +30,23 @@ import org.apache.calcite.sql.SqlJsonConstructorNullClause;
import java.util.LinkedList;
import java.util.List;
-/** Conversion for {@link BuiltInFunctionDefinitions#JSON_OBJECT}. */
+/** Conversion for {@link BuiltInFunctionDefinitions#JSON_ARRAY}. */
@Internal
-class JsonObjectConverter extends CustomizedConverter {
+class JsonArrayConverter extends CustomizedConverter {
@Override
public RexNode convert(CallExpression call, CallExpressionConvertRule.ConvertContext context) {
- checkArgument(call, (call.getChildren().size() - 1) % 2 == 0);
+ checkArgument(call, call.getChildren().size() >= 1);
final List<RexNode> operands = new LinkedList<>();
- final SqlJsonConstructorNullClause onNull =
- ((ValueLiteralExpression) call.getChildren().get(0))
- .getValueAs(JsonOnNull.class)
- .map(this::convertOnNull)
- .orElseThrow(() -> new TableException("Missing argument for ON NULL."));
+ final SqlJsonConstructorNullClause onNull = JsonConverterUtil.getOnNullArgument(call, 0);
operands.add(context.getRelBuilder().getRexBuilder().makeFlag(onNull));
for (int i = 1; i < call.getChildren().size(); i++) {
- final Expression operand = call.getChildren().get(i);
- if (i % 2 == 1 && !(operand instanceof ValueLiteralExpression)) {
- throw new TableException(
- String.format("Argument at position %s must be a string literal.", i));
- }
-
- operands.add(context.toRexNode(operand));
+ operands.add(context.toRexNode(call.getChildren().get(i)));
}
return context.getRelBuilder()
.getRexBuilder()
- .makeCall(FlinkSqlOperatorTable.JSON_OBJECT, operands);
- }
-
- private SqlJsonConstructorNullClause convertOnNull(JsonOnNull onNull) {
- switch (onNull) {
- case NULL:
- return SqlJsonConstructorNullClause.NULL_ON_NULL;
- case ABSENT:
- return SqlJsonConstructorNullClause.ABSENT_ON_NULL;
- default:
- throw new TableException("Unknown ON NULL behavior: " + onNull);
- }
+ .makeCall(FlinkSqlOperatorTable.JSON_ARRAY, operands);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonConverterUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonConverterUtil.java
new file mode 100644
index 0000000..6a9f9b4
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonConverterUtil.java
@@ -0,0 +1,52 @@
+/*
+ * 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.planner.expressions.converter.converters;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.JsonOnNull;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+
+import org.apache.calcite.sql.SqlJsonConstructorNullClause;
+
+/** Utilities for JSON function converters. */
+@Internal
+class JsonConverterUtil {
+ public static SqlJsonConstructorNullClause getOnNullArgument(
+ CallExpression call, int argumentIdx) {
+ return ((ValueLiteralExpression) call.getChildren().get(argumentIdx))
+ .getValueAs(JsonOnNull.class)
+ .map(JsonConverterUtil::convertOnNull)
+ .orElseThrow(() -> new TableException("Missing argument for ON NULL."));
+ }
+
+ private static SqlJsonConstructorNullClause convertOnNull(JsonOnNull onNull) {
+ switch (onNull) {
+ case NULL:
+ return SqlJsonConstructorNullClause.NULL_ON_NULL;
+ case ABSENT:
+ return SqlJsonConstructorNullClause.ABSENT_ON_NULL;
+ default:
+ throw new TableException("Unknown ON NULL behavior: " + onNull);
+ }
+ }
+
+ private JsonConverterUtil() {}
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java
index da1c378..a2dc3a9 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/converters/JsonObjectConverter.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.planner.expressions.converter.converters;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.JsonOnNull;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.Expression;
@@ -42,11 +41,7 @@ class JsonObjectConverter extends CustomizedConverter {
checkArgument(call, (call.getChildren().size() - 1) % 2 == 0);
final List<RexNode> operands = new LinkedList<>();
- final SqlJsonConstructorNullClause onNull =
- ((ValueLiteralExpression) call.getChildren().get(0))
- .getValueAs(JsonOnNull.class)
- .map(this::convertOnNull)
- .orElseThrow(() -> new TableException("Missing argument for ON NULL."));
+ final SqlJsonConstructorNullClause onNull = JsonConverterUtil.getOnNullArgument(call, 0);
operands.add(context.getRelBuilder().getRexBuilder().makeFlag(onNull));
for (int i = 1; i < call.getChildren().size(); i++) {
@@ -63,15 +58,4 @@ class JsonObjectConverter extends CustomizedConverter {
.getRexBuilder()
.makeCall(FlinkSqlOperatorTable.JSON_OBJECT, operands);
}
-
- private SqlJsonConstructorNullClause convertOnNull(JsonOnNull onNull) {
- switch (onNull) {
- case NULL:
- return SqlJsonConstructorNullClause.NULL_ON_NULL;
- case ABSENT:
- return SqlJsonConstructorNullClause.ABSENT_ON_NULL;
- default:
- throw new TableException("Unknown ON NULL behavior: " + onNull);
- }
- }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
index 089de96..02d0bc8 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
@@ -1146,6 +1146,7 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
public static final SqlFunction JSON_VALUE = SqlStdOperatorTable.JSON_VALUE;
public static final SqlFunction JSON_QUERY = SqlStdOperatorTable.JSON_QUERY;
public static final SqlFunction JSON_OBJECT = new SqlJsonObjectFunction();
+ public static final SqlFunction JSON_ARRAY = SqlStdOperatorTable.JSON_ARRAY;
public static final SqlPostfixOperator IS_JSON_VALUE = SqlStdOperatorTable.IS_JSON_VALUE;
public static final SqlPostfixOperator IS_JSON_OBJECT = SqlStdOperatorTable.IS_JSON_OBJECT;
public static final SqlPostfixOperator IS_JSON_ARRAY = SqlStdOperatorTable.IS_JSON_ARRAY;
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 3594bc5..4937ef8 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
@@ -784,6 +784,8 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case JSON_OBJECT => new JsonObjectCallGen(call).generate(ctx, operands, resultType)
+ case JSON_ARRAY => new JsonArrayCallGen(call).generate(ctx, operands, resultType)
+
case _: SqlThrowExceptionFunction =>
val nullValue = generateNullLiteral(resultType, nullCheck = true)
val code =
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
index 0d9a60e..c16378c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.data.{DecimalData, GenericRowData, TimestampData}
import org.apache.flink.table.functions.{ConstantFunctionContext, FunctionContext, UserDefinedFunction}
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.FunctionCodeGenerator.generateFunction
-import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.JSON_OBJECT
+import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.{JSON_ARRAY, JSON_OBJECT}
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.flink.table.planner.utils.Logging
import org.apache.flink.table.types.DataType
@@ -55,6 +55,8 @@ class ExpressionReducer(
private val EMPTY_ROW_TYPE = RowType.of()
private val EMPTY_ROW = new GenericRowData(0)
+ private val nonReducibleJsonFunctions = Seq(JSON_OBJECT, JSON_ARRAY)
+
override def reduce(
rexBuilder: RexBuilder,
constExprs: java.util.List[RexNode],
@@ -81,7 +83,7 @@ class ExpressionReducer(
// Exclude some JSON functions which behave differently when called as an argument of another
// call of one of these functions.
- case (_, call: RexCall) if call.getOperator == JSON_OBJECT => None
+ case (_, call: RexCall) if nonReducibleJsonFunctions.contains(call.getOperator) => None
case (_, e) => Some(e)
}
@@ -146,7 +148,7 @@ class ExpressionReducer(
// if contains python function then just insert the original expression.
reducedValues.add(unreduced)
} else unreduced match {
- case call: RexCall if call.getOperator == JSON_OBJECT =>
+ case call: RexCall if nonReducibleJsonFunctions.contains(call.getOperator) =>
reducedValues.add(unreduced)
case _ =>
unreduced.getType.getSqlTypeName match {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/JsonGenerateUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/JsonGenerateUtils.scala
index fde4b8e..f59aad1 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/JsonGenerateUtils.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/JsonGenerateUtils.scala
@@ -20,7 +20,7 @@ package org.apache.flink.table.planner.codegen
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.planner.codegen.CodeGenUtils.{className, newName, rowFieldReadAccess, typeTerm}
-import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.JSON_OBJECT
+import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.{JSON_ARRAY, JSON_OBJECT}
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
import org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isCharacterString
import org.apache.flink.table.types.logical.LogicalTypeRoot._
@@ -31,15 +31,29 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.{Arr
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.util.RawValue
import org.apache.calcite.rex.{RexCall, RexNode}
+import org.apache.calcite.sql.SqlJsonConstructorNullClause
import java.time.format.DateTimeFormatter
/** Utility for generating JSON function calls. */
object JsonGenerateUtils {
+ /** Returns a term which wraps the given `expression` into a [[JsonNode]]. If the operand
+ * represents another JSON construction function, a raw node is used instead. */
+ def createNodeTerm(
+ ctx: CodeGeneratorContext,
+ containerNodeTerm: String,
+ expression: GeneratedExpression,
+ operand: RexNode): String = {
+ if (isJsonFunctionOperand(operand)) {
+ createRawNodeTerm(containerNodeTerm, expression)
+ } else {
+ createNodeTerm(ctx, containerNodeTerm, expression)
+ }
+ }
+
/**
- * Returns a term which wraps the given <code>valueExpr</code> into a [[JsonNode]] of the
- * appropriate type.
+ * Returns a term which wraps the given `valueExpr` into a [[JsonNode]] of the appropriate type.
*/
def createNodeTerm(
ctx: CodeGeneratorContext,
@@ -100,7 +114,7 @@ object JsonGenerateUtils {
}
/**
- * Returns a term which wraps the given <code>valueExpr</code> as a raw [[JsonNode]].
+ * Returns a term which wraps the given `valueExpr` as a raw [[JsonNode]].
*
* @param containerNodeTerm Name of the [[ContainerNode]] from which to create the raw node.
* @param valueExpr Generated expression of the value which should be wrapped.
@@ -113,6 +127,15 @@ object JsonGenerateUtils {
|""".stripMargin
}
+ /** Convert the operand to [[SqlJsonConstructorNullClause]]. */
+ def getOnNullBehavior(operand: GeneratedExpression): SqlJsonConstructorNullClause = {
+ operand.literalValue match {
+ case Some(onNull: SqlJsonConstructorNullClause) => onNull
+ case _ => throw new CodeGenException(s"Expected operand to be of type"
+ + s"'${typeTerm(classOf[SqlJsonConstructorNullClause])}''")
+ }
+ }
+
/**
* Determines whether the given operand is a call to a JSON function whose result should be
* inserted as a raw value instead of as a character string.
@@ -120,7 +143,7 @@ object JsonGenerateUtils {
def isJsonFunctionOperand(operand: RexNode): Boolean = {
operand match {
case rexCall: RexCall => rexCall.getOperator match {
- case JSON_OBJECT => true
+ case JSON_OBJECT | JSON_ARRAY => true
case _ => false
}
case _ => false
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 82387c9..5086a2b 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
@@ -513,7 +513,4 @@ object BuiltInMethods {
val JSON_QUERY = Types.lookupMethod(classOf[JsonFunctions], "jsonQuery",
classOf[String], classOf[String], classOf[SqlJsonQueryWrapperBehavior],
classOf[SqlJsonQueryEmptyOrErrorBehavior], classOf[SqlJsonQueryEmptyOrErrorBehavior])
-
- val JSON_OBJECT = Types.lookupMethod(classOf[JsonFunctions], "jsonObject",
- classOf[SqlJsonConstructorNullClause], classOf[Array[Any]])
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonArrayCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonArrayCallGen.scala
new file mode 100644
index 0000000..f5275e9
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonArrayCallGen.scala
@@ -0,0 +1,85 @@
+/*
+ * 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.planner.codegen.calls
+
+import org.apache.flink.table.planner.codegen.CodeGenUtils.{BINARY_STRING, className, newName, primitiveTypeTermForType}
+import org.apache.flink.table.planner.codegen.JsonGenerateUtils.{createNodeTerm, getOnNullBehavior}
+import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, GeneratedExpression}
+import org.apache.flink.table.runtime.functions.SqlJsonUtils
+import org.apache.flink.table.types.logical.LogicalType
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.{ArrayNode, NullNode}
+
+import org.apache.calcite.rex.RexCall
+import org.apache.calcite.sql.SqlJsonConstructorNullClause.{ABSENT_ON_NULL, NULL_ON_NULL}
+
+/** [[CallGenerator]] for `JSON_ARRAY`. */
+class JsonArrayCallGen(call: RexCall) extends CallGenerator {
+ private def jsonUtils = className[SqlJsonUtils]
+
+ override def generate(
+ ctx: CodeGeneratorContext,
+ operands: Seq[GeneratedExpression],
+ returnType: LogicalType): GeneratedExpression = {
+
+ val nodeTerm = newName("node")
+ ctx.addReusableMember(s"${className[ArrayNode]} $nodeTerm = $jsonUtils.createArrayNode();")
+
+ val nullNodeTerm = newName("nullNode")
+ ctx.addReusableMember(s"${className[NullNode]} $nullNodeTerm = $nodeTerm.nullNode();")
+
+ val onNull = getOnNullBehavior(operands.head)
+ val populateNodeCode = operands.zipWithIndex.drop(1).map {
+ case (elementExpr, elementIdx) =>
+ val elementTerm = createNodeTerm(ctx, nodeTerm, elementExpr, call.operands.get(elementIdx))
+
+ onNull match {
+ case NULL_ON_NULL =>
+ s"""
+ |if (${elementExpr.nullTerm}) {
+ | $nodeTerm.add($nullNodeTerm);
+ |} else {
+ | $nodeTerm.add($elementTerm);
+ |}
+ |""".stripMargin
+ case ABSENT_ON_NULL =>
+ s"""
+ |if (!${elementExpr.nullTerm}) {
+ | $nodeTerm.add($elementTerm);
+ |}
+ |""".stripMargin
+ }
+ }.mkString
+
+ val resultTerm = newName("result")
+ val resultTermType = primitiveTypeTermForType(returnType)
+ val resultCode =
+ s"""
+ |${operands.map(_.code).mkString}
+ |
+ |$nodeTerm.removeAll();
+ |$populateNodeCode
+ |
+ |$resultTermType $resultTerm =
+ | $BINARY_STRING.fromString($jsonUtils.serializeJson($nodeTerm));
+ |""".stripMargin
+
+ GeneratedExpression(resultTerm, "false", resultCode, returnType)
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonObjectCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonObjectCallGen.scala
index ce09e7b..f946a6d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonObjectCallGen.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonObjectCallGen.scala
@@ -18,20 +18,18 @@
package org.apache.flink.table.planner.codegen.calls
import org.apache.flink.table.planner.codegen.CodeGenUtils._
-import org.apache.flink.table.planner.codegen.JsonGenerateUtils.{createRawNodeTerm, createNodeTerm, generateArrayConverter, generateRowConverter, isJsonFunctionOperand}
-import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGeneratorContext, GeneratedExpression}
+import org.apache.flink.table.planner.codegen.JsonGenerateUtils.{createNodeTerm, getOnNullBehavior}
+import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.runtime.functions.SqlJsonUtils
-import org.apache.flink.table.types.logical.{ArrayType, LogicalType, RowType}
-import org.apache.flink.table.types.logical.LogicalTypeRoot.{ARRAY, MAP, MULTISET, ROW}
+import org.apache.flink.table.types.logical.LogicalType
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.{ArrayNode, NullNode, ObjectNode}
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.{NullNode, ObjectNode}
import org.apache.calcite.rex.RexCall
-import org.apache.calcite.sql.SqlJsonConstructorNullClause
import org.apache.calcite.sql.SqlJsonConstructorNullClause.{ABSENT_ON_NULL, NULL_ON_NULL}
/**
- * [[CallGenerator]] for [[BuiltInMethods.JSON_OBJECT]].
+ * [[CallGenerator]] for `JSON_OBJECT`.
*
* `JSON_OBJECT` returns a character string. However, this creates an issue when nesting calls to
* this function with the intention of creating a nested JSON structure. Instead of a nested JSON
@@ -55,14 +53,10 @@ class JsonObjectCallGen(call: RexCall) extends CallGenerator {
val nullNodeTerm = newName("nullNode")
ctx.addReusableMember(s"${className[NullNode]} $nullNodeTerm = $nodeTerm.nullNode();")
- val onNull = getNullBehavior(operands.head)
+ val onNull = getOnNullBehavior(operands.head)
val populateNodeCode = operands.zipWithIndex.drop(1).grouped(2).map {
case Seq((keyExpr, _), (valueExpr, valueIdx)) =>
- val valueTerm = if (isJsonFunctionOperand(call.operands.get(valueIdx))) {
- createRawNodeTerm(nodeTerm, valueExpr)
- } else {
- createNodeTerm(ctx, nodeTerm, valueExpr)
- }
+ val valueTerm = createNodeTerm(ctx, nodeTerm, valueExpr, call.operands.get(valueIdx))
onNull match {
case NULL_ON_NULL =>
@@ -96,12 +90,4 @@ class JsonObjectCallGen(call: RexCall) extends CallGenerator {
GeneratedExpression(resultTerm, "false", resultCode, returnType)
}
-
- private def getNullBehavior(operand: GeneratedExpression): SqlJsonConstructorNullClause = {
- operand.literalValue match {
- case Some(onNull: SqlJsonConstructorNullClause) => onNull
- case _ => throw new CodeGenException(s"Expected operand to be of type"
- + s"'${typeTerm(classOf[SqlJsonConstructorNullClause])}''")
- }
- }
}
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 d2a8591..6d210cc 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
@@ -57,6 +57,7 @@ import static org.apache.flink.table.api.DataTypes.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;
+import static org.apache.flink.table.api.Expressions.jsonArray;
import static org.apache.flink.table.api.Expressions.jsonObject;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.nullOf;
@@ -79,6 +80,7 @@ public class JsonFunctionsITCase extends BuiltInFunctionTestBase {
testCases.addAll(isJsonSpec());
testCases.addAll(jsonQuerySpec());
testCases.addAll(jsonObjectSpec());
+ testCases.addAll(jsonArraySpec());
return testCases;
}
@@ -531,7 +533,12 @@ public class JsonFunctionsITCase extends BuiltInFunctionTestBase {
"K13",
$("f13"),
"K14",
- jsonObject(JsonOnNull.NULL, "A", "B")),
+ jsonObject(JsonOnNull.NULL, "A", "B"),
+ "K15",
+ jsonArray(
+ JsonOnNull.NULL,
+ "A",
+ jsonObject(JsonOnNull.NULL, "K", "V"))),
"JSON_OBJECT("
+ "'K0' VALUE f0, "
+ "'K1' VALUE f1, "
@@ -547,7 +554,8 @@ public class JsonFunctionsITCase extends BuiltInFunctionTestBase {
+ "'K11' VALUE f11, "
+ "'K12' VALUE f12, "
+ "'K13' VALUE f13, "
- + "'K14' VALUE JSON_OBJECT(KEY 'A' VALUE 'B')"
+ + "'K14' VALUE JSON_OBJECT(KEY 'A' VALUE 'B'), "
+ + "'K15' VALUE JSON_ARRAY('A', JSON_OBJECT('K' VALUE 'V'))"
+ ")",
"{"
+ "\"K0\":\"V\","
@@ -564,12 +572,137 @@ public class JsonFunctionsITCase extends BuiltInFunctionTestBase {
+ "\"K11\":\"VGVzdA==\","
+ "\"K12\":\"VGVzdA==\","
+ "\"K13\":{\"f0\":[{\"f0\":1,\"f1\":2}]},"
- + "\"K14\":{\"A\":\"B\"}"
+ + "\"K14\":{\"A\":\"B\"},"
+ + "\"K15\":[\"A\",{\"K\":\"V\"}]"
+ "}",
STRING().notNull(),
VARCHAR(2000).notNull()));
}
+ private static List<TestSpec> jsonArraySpec() {
+ final Map<String, String> mapData = new HashMap<>();
+ mapData.put("M1", "V1");
+ mapData.put("M2", "V2");
+
+ final Map<String, Integer> multisetData = new HashMap<>();
+ multisetData.put("M1", 1);
+ multisetData.put("M2", 2);
+
+ return Arrays.asList(
+ TestSpec.forFunction(BuiltInFunctionDefinitions.JSON_ARRAY)
+ .onFieldsWithData(0)
+ .testResult(
+ jsonArray(JsonOnNull.NULL),
+ "JSON_ARRAY()",
+ "[]",
+ STRING().notNull(),
+ VARCHAR(2000).notNull())
+ .testResult(
+ jsonArray(JsonOnNull.NULL, nullOf(STRING())),
+ "JSON_ARRAY(CAST(NULL AS STRING) NULL ON NULL)",
+ "[null]",
+ STRING().notNull(),
+ VARCHAR(2000).notNull())
+ .testResult(
+ jsonArray(JsonOnNull.ABSENT, nullOf(STRING())),
+ "JSON_ARRAY(CAST(NULL AS STRING) ABSENT ON NULL)",
+ "[]",
+ STRING().notNull(),
+ VARCHAR(2000).notNull()),
+ TestSpec.forFunction(BuiltInFunctionDefinitions.JSON_ARRAY)
+ .onFieldsWithData(
+ "V",
+ true,
+ 1,
+ 1.23d,
+ 1.23,
+ LocalDateTime.parse("1990-06-02T13:37:42.001"),
+ Instant.parse("1990-06-02T13:37:42.001Z"),
+ Arrays.asList("A1", "A2", "A3"),
+ Row.of("R1", Instant.parse("1990-06-02T13:37:42.001Z")),
+ mapData,
+ multisetData,
+ "Test".getBytes(StandardCharsets.UTF_8),
+ "Test".getBytes(StandardCharsets.UTF_8),
+ Row.of(Collections.singletonList(Row.of(1, 2))))
+ .andDataTypes(
+ STRING(),
+ BOOLEAN(),
+ INT(),
+ DOUBLE(),
+ DECIMAL(3, 2),
+ TIMESTAMP(3),
+ TIMESTAMP_WITH_LOCAL_TIME_ZONE(3),
+ ARRAY(STRING()),
+ ROW(STRING(), TIMESTAMP_WITH_LOCAL_TIME_ZONE(3)),
+ MAP(STRING(), STRING()),
+ MAP(STRING(), INT()),
+ BINARY(4),
+ VARBINARY(4),
+ ROW(ARRAY(ROW(INT(), INT()))))
+ .withFunction(CreateMultiset.class)
+ .testResult(
+ jsonArray(
+ JsonOnNull.NULL,
+ $("f0"),
+ $("f1"),
+ $("f2"),
+ $("f3"),
+ $("f4"),
+ $("f5"),
+ $("f6"),
+ $("f7"),
+ $("f8"),
+ $("f9"),
+ call("CreateMultiset", $("f10")),
+ $("f11"),
+ $("f12"),
+ $("f13"),
+ jsonArray(JsonOnNull.NULL, "V"),
+ jsonObject(
+ JsonOnNull.NULL,
+ "K",
+ jsonArray(JsonOnNull.NULL, "V"))),
+ "JSON_ARRAY("
+ + "f0, "
+ + "f1, "
+ + "f2, "
+ + "f3, "
+ + "f4, "
+ + "f5, "
+ + "f6, "
+ + "f7, "
+ + "f8, "
+ + "f9, "
+ + "CreateMultiset(f10), "
+ + "f11, "
+ + "f12, "
+ + "f13, "
+ + "JSON_ARRAY('V'), "
+ + "JSON_OBJECT('K' VALUE JSON_ARRAY('V'))"
+ + ")",
+ "["
+ + "\"V\","
+ + "true,"
+ + "1,"
+ + "1.23,"
+ + "1.23,"
+ + "\"1990-06-02T13:37:42.001\","
+ + "\"1990-06-02T13:37:42.001Z\","
+ + "[\"A1\",\"A2\",\"A3\"],"
+ + "{\"f0\":\"R1\",\"f1\":\"1990-06-02T13:37:42.001Z\"},"
+ + "{\"M1\":\"V1\",\"M2\":\"V2\"},"
+ + "{\"M1\":1,\"M2\":2},"
+ + "\"VGVzdA==\","
+ + "\"VGVzdA==\","
+ + "{\"f0\":[{\"f0\":1,\"f1\":2}]},"
+ + "[\"V\"],"
+ + "{\"K\":[\"V\"]}"
+ + "]",
+ STRING().notNull(),
+ VARCHAR(2000).notNull()));
+ }
+
// ---------------------------------------------------------------------------------------------
/**
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 a7bbf4c..b942486 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
@@ -25,6 +25,7 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
/**
@@ -43,6 +44,11 @@ public class SqlJsonUtils {
return MAPPER.createObjectNode();
}
+ /** Returns a new {@link ArrayNode}. */
+ public static ArrayNode createArrayNode() {
+ return MAPPER.createArrayNode();
+ }
+
/** Serializes the given {@link JsonNode} to a JSON string. */
public static String serializeJson(JsonNode node) {
try {