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
commit 469332bf2947919049b95982f89e78e35a471a8a Author: Ramin Gharib <[email protected]> AuthorDate: Wed Oct 28 17:08:30 2026 +0100 [FLINK-39049][table] Add VARIANT_GET and TRY_VARIANT_GET `VARIANT_GET(variant, path[, returnType])` returns the value of a Variant at a path, cast to `returnType`, or as a Variant without `returnType`. `TRY_VARIANT_GET` returns NULL instead of an error when the value cannot be cast or a path that is not a literal is malformed. Paths follow FLIP-618: `$`, `.name`, `['name']`, `["name"]` and the 0-based `[n]`, without wildcards, filters, slices, negative indices or path modes. The internal `SimplePath` in flink-table-common parses them, so a malformed literal path fails at planning. A missing field, an index past the end, or a step on the wrong kind of value gives NULL. `returnType` is a string literal, as in Spark, parsed with `LogicalTypeParser`. The cast reuses `FromVariantConverter`, so `VARIANT_GET(v, p, 'T')` casts like `CAST`, and a variant null gives NULL unless `returnType` is `VARIANT`. The result is always nullable. The Table API and PyFlink get `variantGet` / `variant_get` and `tryVariantGet` / `try_variant_get`. `getLocalTimeZone(ReadableConfig)` moves from the planner's `TableConfigUtils` to `TableConfigValidation` in flink-table-api-java, so the runtime function can read the session time zone. The planner's method delegates to it. The docs add the function entries and a "Variant paths" section. They also correct `variant[INT]` and `variant[STRING]`, which throw on the wrong kind of value instead of returning NULL. Generated-by: Claude Code 2.1.310 (Claude Opus 5.5) --- .../docs/sql/functions/built-in-functions.md | 30 ++ docs/content.zh/docs/sql/reference/data-types.md | 5 +- .../docs/sql/functions/built-in-functions.md | 30 ++ docs/content/docs/sql/reference/data-types.md | 5 +- docs/data/sql_functions.yml | 64 ++++- docs/data/sql_functions_zh.yml | 62 ++++- .../docs/reference/pyflink.table/expressions.rst | 11 + flink-python/pyflink/table/expression.py | 34 +++ .../pyflink/table/tests/test_expression.py | 8 + .../flink/table/api/internal/BaseExpressions.java | 66 +++++ .../table/api/internal/TableConfigValidation.java | 17 ++ .../functions/BuiltInFunctionDefinitions.java | 20 ++ .../strategies/SpecificInputTypeStrategies.java | 3 + .../strategies/SpecificTypeStrategies.java | 3 + .../strategies/VariantGetInputTypeStrategy.java | 109 ++++++++ .../strategies/VariantGetTypeStrategy.java | 89 ++++++ .../org/apache/flink/table/utils/SimplePath.java | 303 +++++++++++++++++++++ .../apache/flink/table/utils/SimplePathTest.java | 114 ++++++++ .../table/planner/utils/TableConfigUtils.java | 10 +- .../planner/functions/VariantFunctionsITCase.java | 167 ++++++++++++ .../nodes/exec/stream/VariantSemanticTest.java | 52 ++++ .../functions/scalar/TryVariantGetFunction.java | 36 +++ .../functions/scalar/VariantGetFunction.java | 182 +++++++++++++ 23 files changed, 1404 insertions(+), 16 deletions(-) diff --git a/docs/content.zh/docs/sql/functions/built-in-functions.md b/docs/content.zh/docs/sql/functions/built-in-functions.md index 79115623e0a..ad1e9f83a7b 100644 --- a/docs/content.zh/docs/sql/functions/built-in-functions.md +++ b/docs/content.zh/docs/sql/functions/built-in-functions.md @@ -233,6 +233,36 @@ A document fails to parse if it: In all of these cases, `TRY_PARSE_XML` returns `NULL`. +#### Variant paths + +`VARIANT_GET` and `TRY_VARIANT_GET` take a path that names one place in a Variant. A path starts +with `$`, the whole value, followed by steps: `.name`, `['name']` or `["name"]` for a field, and +`[n]` for an array element. For example, on +`PARSE_JSON('{"user": {"name": "ada"}, "items": [{"qty": 2}]}')`: + +| Path | Points at | +|---------------------|------------------------------------| +| `$` | the whole value | +| `$.user.name` | `"ada"` | +| `$['user']["name"]` | `"ada"` | +| `$.items[0].qty` | `2` | +| `$.items[1]` | nothing, the index is past the end | +| `$.user[0]` | nothing, `user` is not an array | + +- Array indices start at 0, as in `JSON_VALUE`, while `variant[INT]` starts at 1. So + `VARIANT_GET(v, '$.items[0]')` and `v['items'][1]` are the same element. +- Use brackets with single or double quotes for a name with whitespace or one of the characters + `. [ ] ( ) * ' " \`, for example `$['first name']` or `$["first name"]`. Inside the quotes, `\\`, + `\'` and `\"` stand for a backslash and a quote. +- Wildcards, filters, slices, negative indices and the `lax` and `strict` modes are not supported, + so a path always names one place. +- Trailing spaces are ignored. +- A malformed path literal fails at planning time. Any other malformed path fails when the function + runs. + +A path points at nothing if a field is missing, an index is past the end, or a step does not match +the kind of value, such as a field of an array. Each function describes what it does in that case. + ### 值构建函数 {{< sql_functions_zh "valueconstruction" >}} diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 703185f11de..fdfc3a12cde 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1551,7 +1551,10 @@ To store one of these values, cast the number to a `VARIANT` instead of parsing A `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. A cast succeeds only when the target holds the stored value without reinterpreting it, so a value is never wrapped or rounded -to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. +to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. `VARIANT_GET` and +`TRY_VARIANT_GET` read the value at a +[path]({{< ref "docs/sql/functions/built-in-functions" >}}#variant-paths) such as `'$.items[0].qty'` +and cast it by the same rules, returning `NULL` when nothing is at the path. | Stored kind | Succeeds for | |-----------------|-----------------------------------------------------| diff --git a/docs/content/docs/sql/functions/built-in-functions.md b/docs/content/docs/sql/functions/built-in-functions.md index 01bf2070579..0d8ebd0abb4 100644 --- a/docs/content/docs/sql/functions/built-in-functions.md +++ b/docs/content/docs/sql/functions/built-in-functions.md @@ -236,6 +236,36 @@ A document fails to parse if it: In all of these cases, `TRY_PARSE_XML` returns `NULL`. +#### Variant paths + +`VARIANT_GET` and `TRY_VARIANT_GET` take a path that names one place in a Variant. A path starts +with `$`, the whole value, followed by steps: `.name`, `['name']` or `["name"]` for a field, and +`[n]` for an array element. For example, on +`PARSE_JSON('{"user": {"name": "ada"}, "items": [{"qty": 2}]}')`: + +| Path | Points at | +|---------------------|------------------------------------| +| `$` | the whole value | +| `$.user.name` | `"ada"` | +| `$['user']["name"]` | `"ada"` | +| `$.items[0].qty` | `2` | +| `$.items[1]` | nothing, the index is past the end | +| `$.user[0]` | nothing, `user` is not an array | + +- Array indices start at 0, as in `JSON_VALUE`, while `variant[INT]` starts at 1. So + `VARIANT_GET(v, '$.items[0]')` and `v['items'][1]` are the same element. +- Use brackets with single or double quotes for a name with whitespace or one of the characters + `. [ ] ( ) * ' " \`, for example `$['first name']` or `$["first name"]`. Inside the quotes, `\\`, + `\'` and `\"` stand for a backslash and a quote. +- Wildcards, filters, slices, negative indices and the `lax` and `strict` modes are not supported, + so a path always names one place. +- Trailing spaces are ignored. +- A malformed path literal fails at planning time. Any other malformed path fails when the function + runs. + +A path points at nothing if a field is missing, an index is past the end, or a step does not match +the kind of value, such as a field of an array. Each function describes what it does in that case. + ### Value Construction Functions {{< sql_functions "valueconstruction" >}} diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index 1ac289a0402..3cedce4ff8c 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1559,7 +1559,10 @@ To store one of these values, cast the number to a `VARIANT` instead of parsing A `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. A cast succeeds only when the target holds the stored value without reinterpreting it, so a value is never wrapped or rounded -to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. +to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. `VARIANT_GET` and +`TRY_VARIANT_GET` read the value at a +[path]({{< ref "docs/sql/functions/built-in-functions" >}}#variant-paths) such as `'$.items[0].qty'` +and cast it by the same rules, returning `NULL` when nothing is at the path. | Stored kind | Succeeds for | |-----------------|-----------------------------------------------------| diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml index 22463dcdf00..fadf6f734b4 100644 --- a/docs/data/sql_functions.yml +++ b/docs/data/sql_functions.yml @@ -1451,13 +1451,69 @@ variant: - sql: variant '[' INT ']' table: VARIANT.at(INT) description: | - If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at - the specified index. The index starts from 1, If the index is out of range, it returns NULL. + If the VARIANT is an array, returns a VARIANT holding the element at the index. The index + starts at 1. If the index is out of range, returns NULL. If the VARIANT is not an array, an + error is thrown. To return NULL instead, use `VARIANT_GET`. - sql: variant '[' STRING ']' table: VARIANT.at(STRING) description: | - If the VARIANT is a MAP value that has an element with this key, a VARIANT holding - the associated value is returned. Otherwise, NULL is returned. + If the VARIANT is an object that has a field with this key, returns a VARIANT holding the + value of the field. If the object has no such field, returns NULL. If the VARIANT is not an + object, an error is thrown. To return NULL instead, use `VARIANT_GET`. + - sql: VARIANT_GET(variant, path[, returnType]) + table: VARIANT.variantGet(path[, returnType]) + description: | + Returns the value of a Variant at a path, cast to `returnType`. Without `returnType`, the + value is returned as a Variant. If the value cannot be cast to `returnType`, an error is + thrown. To return NULL instead of an error, use the `TRY_VARIANT_GET` function. See the + "Variant paths" section below the function table for the path syntax. + + Returns NULL if: + + * `variant` or `path` is NULL, + * nothing is at the path, or + * the value at the path is a variant null and `returnType` is not `'VARIANT'`. + + Without `returnType`, or with `'VARIANT'`, a variant null stays a variant null. It is not SQL + NULL, so `IS NULL` is false and `JSON_STRING` returns `'null'`. + + `returnType` is a string literal such as `'INT'` or `'ARRAY<ROW<sku STRING, qty INT>>'`, and + must not be NULL. The value is cast the way `CAST(v AS returnType)` casts a Variant. The + result is always nullable. + + ```sql + -- 17 of type INT + VARIANT_GET(PARSE_JSON('{"order_id": 17}'), '$.order_id', 'INT') + + -- 2 of type INT + VARIANT_GET(PARSE_JSON('{"items": [{"qty": 2}]}'), '$.items[0].qty', 'INT') + + -- NULL of type STRING, nothing is at the path + VARIANT_GET(PARSE_JSON('{"order_id": 17}'), '$.user.name', 'STRING') + + -- NULL of type INT, a variant null cast to INT + VARIANT_GET(PARSE_JSON('{"coupon": null}'), '$.coupon', 'INT') + + -- null of type VARIANT, a variant null and not SQL NULL + VARIANT_GET(PARSE_JSON('{"coupon": null}'), '$.coupon') + + -- {"name":"ada"} of type VARIANT + VARIANT_GET(PARSE_JSON('{"user": {"name": "ada"}}'), '$.user') + + -- fails, a Variant string does not cast to INT + VARIANT_GET(PARSE_JSON('{"price": "12.50"}'), '$.price', 'INT') + ``` + - sql: TRY_VARIANT_GET(variant, path[, returnType]) + table: VARIANT.tryVariantGet(path[, returnType]) + description: | + Returns the value of a Variant at a path like `VARIANT_GET`, but returns NULL instead of an + error if the value cannot be cast to `returnType`, or if a path that is not a literal is + malformed. A malformed path literal still fails at planning time. + + ```sql + -- NULL of type INT, a Variant string does not cast to INT + TRY_VARIANT_GET(PARSE_JSON('{"price": "12.50"}'), '$.price', 'INT') + ``` valueconstruction: - sql: | diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml index d1dc90163fb..c9b379fcbe2 100644 --- a/docs/data/sql_functions_zh.yml +++ b/docs/data/sql_functions_zh.yml @@ -1549,12 +1549,70 @@ variant: - sql: variant '[' INT ']' table: VARIANT.at(INT) description: | - 如果这是一个 ARRAY 类型的 VARIANT，则返回一个 VARIANT，其值为指定索引处的元素。索引从 1 开始，如果索引超出范围，则返回 NULL。 + If the VARIANT is an array, returns a VARIANT holding the element at the index. The index + starts at 1. If the index is out of range, returns NULL. If the VARIANT is not an array, an + error is thrown. To return NULL instead, use `VARIANT_GET`. - sql: variant '[' STRING ']' table: VARIANT.at(STRING) description: | - 如果这是一个 MAP 类型的 VARIANT，则返回一个 VARIANT，其值为指定 key 对应的值。否则返回 NULL。 + If the VARIANT is an object that has a field with this key, returns a VARIANT holding the + value of the field. If the object has no such field, returns NULL. If the VARIANT is not an + object, an error is thrown. To return NULL instead, use `VARIANT_GET`. + - sql: VARIANT_GET(variant, path[, returnType]) + table: VARIANT.variantGet(path[, returnType]) + description: | + Returns the value of a Variant at a path, cast to `returnType`. Without `returnType`, the + value is returned as a Variant. If the value cannot be cast to `returnType`, an error is + thrown. To return NULL instead of an error, use the `TRY_VARIANT_GET` function. See the + "Variant paths" section below the function table for the path syntax. + + Returns NULL if: + + * `variant` or `path` is NULL, + * nothing is at the path, or + * the value at the path is a variant null and `returnType` is not `'VARIANT'`. + + Without `returnType`, or with `'VARIANT'`, a variant null stays a variant null. It is not SQL + NULL, so `IS NULL` is false and `JSON_STRING` returns `'null'`. + + `returnType` is a string literal such as `'INT'` or `'ARRAY<ROW<sku STRING, qty INT>>'`, and + must not be NULL. The value is cast the way `CAST(v AS returnType)` casts a Variant. The + result is always nullable. + + ```sql + -- 17 of type INT + VARIANT_GET(PARSE_JSON('{"order_id": 17}'), '$.order_id', 'INT') + + -- 2 of type INT + VARIANT_GET(PARSE_JSON('{"items": [{"qty": 2}]}'), '$.items[0].qty', 'INT') + + -- NULL of type STRING, nothing is at the path + VARIANT_GET(PARSE_JSON('{"order_id": 17}'), '$.user.name', 'STRING') + + -- NULL of type INT, a variant null cast to INT + VARIANT_GET(PARSE_JSON('{"coupon": null}'), '$.coupon', 'INT') + + -- null of type VARIANT, a variant null and not SQL NULL + VARIANT_GET(PARSE_JSON('{"coupon": null}'), '$.coupon') + + -- {"name":"ada"} of type VARIANT + VARIANT_GET(PARSE_JSON('{"user": {"name": "ada"}}'), '$.user') + + -- fails, a Variant string does not cast to INT + VARIANT_GET(PARSE_JSON('{"price": "12.50"}'), '$.price', 'INT') + ``` + - sql: TRY_VARIANT_GET(variant, path[, returnType]) + table: VARIANT.tryVariantGet(path[, returnType]) + description: | + Returns the value of a Variant at a path like `VARIANT_GET`, but returns NULL instead of an + error if the value cannot be cast to `returnType`, or if a path that is not a literal is + malformed. A malformed path literal still fails at planning time. + + ```sql + -- NULL of type INT, a Variant string does not cast to INT + TRY_VARIANT_GET(PARSE_JSON('{"price": "12.50"}'), '$.price', 'INT') + ``` valueconstruction: diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst index e2ee9777c58..6c09c999e1c 100644 --- a/flink-python/docs/reference/pyflink.table/expressions.rst +++ b/flink-python/docs/reference/pyflink.table/expressions.rst @@ -335,6 +335,17 @@ JSON functions Expression.json_length Expression.json_type +variant functions +----------------- + +.. currentmodule:: pyflink.table.expression + +.. autosummary:: + :toctree: api/ + + Expression.variant_get + Expression.try_variant_get + value modification functions ---------------------------- diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py index 426a3059bd7..e497ecb5590 100644 --- a/flink-python/pyflink/table/expression.py +++ b/flink-python/pyflink/table/expression.py @@ -2456,6 +2456,40 @@ class Expression(Generic[T]): else: return _binary_op("tryParseXml")(self, force_array) + def variant_get(self, path, return_type: DataType = None) -> 'Expression': + """ + Returns the value of a VARIANT at a path, cast to return_type. Without return_type, the + value is returned as a VARIANT. If the VARIANT or the path is NULL, or nothing is at the + path, NULL is returned. A variant null at the path gives NULL, unless return_type is + omitted or DataTypes.VARIANT(), which keeps it as a variant null. If the value cannot be + cast to return_type, an error is thrown. To return NULL instead, use + :func:`~Expression.try_variant_get`. + + Array indices in a path such as '$.items[0].qty' start at 0. See the SQL function + VARIANT_GET for the path syntax. + + Example: + :: + + >>> lit('{"a": {"b": 1}}').parse_json().variant_get('$.a.b', DataTypes.INT()) # 1 + >>> lit('{"a": 1}').parse_json().variant_get('$.b', DataTypes.INT()) # NULL + """ + if return_type is None: + return _binary_op("variantGet")(self, path) + else: + return _ternary_op("variantGet")(self, path, _to_java_data_type(return_type)) + + def try_variant_get(self, path, return_type: DataType = None) -> 'Expression': + """ + Returns the value of a VARIANT at a path, cast to return_type. Unlike + :func:`~Expression.variant_get`, NULL is returned instead of an error if the value cannot + be cast to return_type or a path that is not a literal is malformed. + """ + if return_type is None: + return _binary_op("tryVariantGet")(self, path) + else: + return _ternary_op("tryVariantGet")(self, path, _to_java_data_type(return_type)) + def json_length(self, path=None) -> 'Expression': """ Returns the number of elements in a JSON document, or the length of the value at the diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py index 08c4f9b9d5f..fba77f6b89e 100644 --- a/flink-python/pyflink/table/tests/test_expression.py +++ b/flink-python/pyflink/table/tests/test_expression.py @@ -280,6 +280,14 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase): self.assertEqual("PARSE_XML(a, b)", str(expr1.parse_xml(expr2))) self.assertEqual("TRY_PARSE_XML(a, b)", str(expr1.try_parse_xml(expr2))) + # variant functions + self.assertEqual("VARIANT_GET(a, '$.x')", str(expr1.variant_get('$.x'))) + self.assertEqual("VARIANT_GET(a, '$.x', 'INT')", + str(expr1.variant_get('$.x', DataTypes.INT()))) + self.assertEqual("TRY_VARIANT_GET(a, b)", str(expr1.try_variant_get(expr2))) + self.assertEqual("TRY_VARIANT_GET(a, '$.x', 'ARRAY<INT>')", + str(expr1.try_variant_get('$.x', DataTypes.ARRAY(DataTypes.INT())))) + # bitmap functions self.assertEqual("BITMAP_AND(a, b)", str(expr1.bitmap_and(expr2))) self.assertEqual("BITMAP_ANDNOT(a, b)", str(expr1.bitmap_andnot(expr2))) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java index 29f12604fc8..a5d041aca8e 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java @@ -239,10 +239,12 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRUNCA import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRY_CAST; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRY_PARSE_JSON; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRY_PARSE_XML; +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRY_VARIANT_GET; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.UNHEX; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.UPPER; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.URL_DECODE; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.URL_ENCODE; +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.VARIANT_GET; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.VAR_POP; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.VAR_SAMP; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.WINDOW_END; @@ -1481,6 +1483,70 @@ public abstract class BaseExpressions<InType, OutType> { unresolvedCall(TRY_PARSE_XML, toExpr(), objectToExpression(forceArray))); } + /** + * Returns the value of a {@link DataTypes#VARIANT()} at a path as a {@link + * DataTypes#VARIANT()}. If the variant or the path is {@code NULL}, or nothing is at the path, + * {@code NULL} is returned. A variant null at the path stays a variant null, which is not + * {@code NULL}. + * + * <p>This is a shortcut for {@code variantGet(path, DataTypes.VARIANT())}. See {@link + * #variantGet(Object, DataType)}. + */ + public OutType variantGet(InType path) { + return toApiSpecificExpression( + unresolvedCall(VARIANT_GET, toExpr(), objectToExpression(path))); + } + + /** + * Returns the value of a {@link DataTypes#VARIANT()} at a path such as {@code $.items[0].qty}, + * cast to the given type. Array indices start at 0. If the variant or the path is {@code NULL}, + * or nothing is at the path, {@code NULL} is returned. If the value cannot be cast to the type, + * an error is thrown. To return {@code NULL} instead, use {@link #tryVariantGet(Object, + * DataType)}. + * + * <p>The value is cast like {@code CAST}, so a variant null gives {@code NULL} unless the type + * is {@link DataTypes#VARIANT()}. See the SQL function {@code VARIANT_GET} for the path syntax. + * + * @param returnType the type of the result, always nullable. + */ + public OutType variantGet(InType path, DataType returnType) { + return toApiSpecificExpression( + unresolvedCall( + VARIANT_GET, + toExpr(), + objectToExpression(path), + valueLiteral(returnType.getLogicalType().asSerializableString()))); + } + + /** + * Returns the value of a {@link DataTypes#VARIANT()} at a path as a {@link + * DataTypes#VARIANT()}. If nothing is at the path, or a path that is not a literal is + * malformed, {@code NULL} is returned. + * + * <p>This is a shortcut for {@code tryVariantGet(path, DataTypes.VARIANT())}. See {@link + * #tryVariantGet(Object, DataType)}. + */ + public OutType tryVariantGet(InType path) { + return toApiSpecificExpression( + unresolvedCall(TRY_VARIANT_GET, toExpr(), objectToExpression(path))); + } + + /** + * Returns the value of a {@link DataTypes#VARIANT()} at a path, cast to the given type. Unlike + * {@link #variantGet(Object, DataType)}, {@code NULL} is returned instead of an error if the + * value cannot be cast to the type or a path that is not a literal is malformed. + * + * @param returnType the type of the result, always nullable. + */ + public OutType tryVariantGet(InType path, DataType returnType) { + return toApiSpecificExpression( + unresolvedCall( + TRY_VARIANT_GET, + toExpr(), + objectToExpression(path), + valueLiteral(returnType.getLogicalType().asSerializableString()))); + } + /** Returns the base string decoded with base64. */ public OutType fromBase64() { return toApiSpecificExpression(unresolvedCall(FROM_BASE64, toExpr())); diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java index b6d1eae5f02..777a8df9d58 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java @@ -19,8 +19,10 @@ package org.apache.flink.table.api.internal; import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.config.TableConfigOptions; import java.time.ZoneId; @@ -49,6 +51,21 @@ public class TableConfigValidation { } } + /** + * Similar to {@link TableConfig#getLocalTimeZone()} but extracting it from a generic {@link + * ReadableConfig}. + * + * @see TableConfig#getLocalTimeZone() + */ + public static ZoneId getLocalTimeZone(ReadableConfig config) { + final String zone = config.get(TableConfigOptions.LOCAL_TIME_ZONE); + if (TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)) { + return ZoneId.systemDefault(); + } + validateTimeZone(zone); + return ZoneId.of(zone); + } + private TableConfigValidation() { // No instantiation } 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 6c09b633d81..31ab26c0401 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 @@ -3289,6 +3289,26 @@ public final class BuiltInFunctionDefinitions { .runtimeProvided() .build(); + public static final BuiltInFunctionDefinition VARIANT_GET = + BuiltInFunctionDefinition.newBuilder() + .name("VARIANT_GET") + .kind(SCALAR) + .inputTypeStrategy(SpecificInputTypeStrategies.VARIANT_GET) + .outputTypeStrategy(SpecificTypeStrategies.VARIANT_GET) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.VariantGetFunction") + .build(); + + public static final BuiltInFunctionDefinition TRY_VARIANT_GET = + BuiltInFunctionDefinition.newBuilder() + .name("TRY_VARIANT_GET") + .kind(SCALAR) + .inputTypeStrategy(SpecificInputTypeStrategies.VARIANT_GET) + .outputTypeStrategy(SpecificTypeStrategies.VARIANT_GET) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.TryVariantGetFunction") + .build(); + // -------------------------------------------------------------------------------------------- // Bitmap 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 e012f310953..ea5f2433a12 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 @@ -66,6 +66,9 @@ public final class SpecificInputTypeStrategies { /** See {@link ObjectOfInputTypeStrategy}. */ public static final InputTypeStrategy OBJECT_OF = new ObjectOfInputTypeStrategy(); + /** See {@link VariantGetInputTypeStrategy}. */ + public static final InputTypeStrategy VARIANT_GET = new VariantGetInputTypeStrategy(); + /** See {@link ObjectUpdateInputTypeStrategy}. */ public static final InputTypeStrategy OBJECT_UPDATE = new ObjectUpdateInputTypeStrategy(); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java index 00a82abb95d..cf04745381b 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java @@ -231,6 +231,9 @@ public final class SpecificTypeStrategies { /** Type strategy specific for {@link BuiltInFunctionDefinitions#OBJECT_UPDATE}. */ public static final TypeStrategy OBJECT_UPDATE = new ObjectUpdateTypeStrategy(); + /** See {@link VariantGetTypeStrategy}. */ + public static final TypeStrategy VARIANT_GET = new VariantGetTypeStrategy(); + /** Type strategy specific for {@link BuiltInFunctionDefinitions#ML_PREDICT}. */ public static final TypeStrategy ML_PREDICT_OUTPUT_TYPE_STRATEGY = MLPredictTypeStrategy.ML_PREDICT_OUTPUT_TYPE_STRATEGY; diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetInputTypeStrategy.java new file mode 100644 index 00000000000..6ffd02c00d4 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetInputTypeStrategy.java @@ -0,0 +1,109 @@ +/* + * 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.types.inference.strategies; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.functions.FunctionDefinition; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.inference.ArgumentCount; +import org.apache.flink.table.types.inference.CallContext; +import org.apache.flink.table.types.inference.InputTypeStrategy; +import org.apache.flink.table.types.inference.Signature; +import org.apache.flink.table.types.logical.LogicalTypeFamily; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.utils.SimplePath; + +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.table.types.inference.InputTypeStrategies.LITERAL; +import static org.apache.flink.table.types.inference.InputTypeStrategies.and; +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.sequence; +import static org.apache.flink.table.types.inference.strategies.VariantGetTypeStrategy.ARG_RETURN_TYPE; + +/** + * Input strategy for {@code VARIANT_GET(variant, path [, returnType])} and {@code TRY_VARIANT_GET}. + * + * <p>A literal path must be a {@link SimplePath}, so a malformed one fails at planning time. A path + * that is not a literal is parsed when the function runs. The {@code returnType} must be a literal + * that names a type {@code VARIANT} can be cast to, see {@link VariantGetTypeStrategy}. + */ +@Internal +public final class VariantGetInputTypeStrategy implements InputTypeStrategy { + + private static final int ARG_PATH = 1; + + private static final InputTypeStrategy SIGNATURES = + or( + sequence( + List.of("variant", "path"), + List.of( + logical(LogicalTypeRoot.VARIANT), + logical(LogicalTypeFamily.CHARACTER_STRING))), + sequence( + List.of("variant", "path", "returnType"), + List.of( + logical(LogicalTypeRoot.VARIANT), + logical(LogicalTypeFamily.CHARACTER_STRING), + and(logical(LogicalTypeFamily.CHARACTER_STRING), LITERAL)))); + + @Override + public ArgumentCount getArgumentCount() { + return SIGNATURES.getArgumentCount(); + } + + @Override + public Optional<List<DataType>> inferInputTypes( + final CallContext callContext, final boolean throwOnFailure) { + final Optional<List<DataType>> inferredDataTypes = + SIGNATURES.inferInputTypes(callContext, throwOnFailure); + if (inferredDataTypes.isEmpty()) { + return inferredDataTypes; + } + + if (callContext.isArgumentLiteral(ARG_PATH)) { + final Optional<String> path = callContext.getArgumentValue(ARG_PATH, String.class); + if (path.isPresent()) { + try { + SimplePath.parse(path.get()); + } catch (ValidationException e) { + return callContext.fail(throwOnFailure, e.getMessage()); + } + } + } + + if (callContext.getArgumentDataTypes().size() > ARG_RETURN_TYPE) { + try { + VariantGetTypeStrategy.resolveTargetType(callContext); + } catch (ValidationException e) { + return callContext.fail(throwOnFailure, e.getMessage()); + } + } + + return inferredDataTypes; + } + + @Override + public List<Signature> getExpectedSignatures(final FunctionDefinition definition) { + return SIGNATURES.getExpectedSignatures(definition); + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetTypeStrategy.java new file mode 100644 index 00000000000..7ddf00a2594 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VariantGetTypeStrategy.java @@ -0,0 +1,89 @@ +/* + * 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.types.inference.strategies; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.inference.CallContext; +import org.apache.flink.table.types.inference.TypeStrategy; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.VariantType; +import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; +import org.apache.flink.table.types.logical.utils.LogicalTypeParser; + +import java.util.Optional; + +import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType; + +/** + * Type strategy for {@code VARIANT_GET(variant, path [, returnType])} and {@code TRY_VARIANT_GET}. + * + * <p>The result is the type named by the {@code returnType} literal, or {@code VARIANT} without + * one. It is always nullable, because a path that points at nothing gives {@code NULL}. + */ +@Internal +public final class VariantGetTypeStrategy implements TypeStrategy { + + static final int ARG_RETURN_TYPE = 2; + + @Override + public Optional<DataType> inferType(final CallContext callContext) { + if (callContext.getArgumentDataTypes().size() <= ARG_RETURN_TYPE) { + return Optional.of(DataTypes.VARIANT().nullable()); + } + return Optional.of(resolveTargetType(callContext)); + } + + /** + * Resolves the {@code returnType} literal to a nullable type that {@code VARIANT} can be cast + * to. + * + * @throws ValidationException if the literal is not a type or names a type that {@code VARIANT} + * cannot be cast to + */ + static DataType resolveTargetType(final CallContext callContext) { + // The input strategy only accepts a non-null literal. + final String typeString = + callContext + .getArgumentValue(ARG_RETURN_TYPE, String.class) + .orElseThrow(IllegalStateException::new); + final LogicalType logicalType; + try { + logicalType = + LogicalTypeParser.parse( + typeString, callContext.getDataTypeFactory().getClassLoader()) + .copy(true); + } catch (RuntimeException e) { + throw new ValidationException( + String.format( + "Invalid type '%s' for %s: %s", + typeString, callContext.getName(), e.getMessage()), + e); + } + if (!LogicalTypeCasts.supportsExplicitCast(new VariantType(), logicalType)) { + throw new ValidationException( + String.format( + "%s cannot return type %s, because a VARIANT cannot be cast to it.", + callContext.getName(), logicalType.asSummaryString())); + } + return fromLogicalToDataType(logicalType); + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/SimplePath.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/SimplePath.java new file mode 100644 index 00000000000..5b6204cad5e --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/SimplePath.java @@ -0,0 +1,303 @@ +/* + * 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.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.regex.Pattern; + +/** + * A path that points at exactly one place in a value, such as {@code $.user.email} or {@code + * $['items'][0]}. + * + * <p>The syntax is the part of the JSON functions' path syntax that names a single place: no + * wildcards, filters, slices, negative indices, functions or {@code lax} and {@code strict} modes. + * Every path that this class accepts means the same in the JSON functions. A path starts with + * {@code $}, the whole value, followed by any number of steps: + * + * <ul> + * <li>{@code .name} is a field. The name cannot contain whitespace or one of the characters + * {@code . [ ] ( ) * ' " \}. + * <li>{@code ['name']} and {@code ["name"]} are a field with any name. Inside the quotes, a + * backslash followed by a backslash or a quote stands for that character. + * <li>{@code [n]} is the array element at the 0-based index {@code n}. + * </ul> + * + * <p>Trailing spaces are ignored, as in the JSON functions, so a space-padded {@code CHAR(n)} path + * works. + */ +@Internal +public final class SimplePath implements Serializable { + + private static final long serialVersionUID = 1L; + + private static final String SYNTAX = + "A path is '$' followed by the steps .name, ['name'], [\"name\"] or [index], " + + "without wildcards, filters, slices, negative indices or functions. " + + "Use brackets for a name that has whitespace or one of the characters " + + ". [ ] ( ) * ' \" \\ in it."; + + private static final String NAME_STOP_CHARACTERS = ".[]()*'\"\\"; + + private static final Pattern PATH_MODE_PREFIX = + Pattern.compile("\\s*(strict|lax)\\s.*", Pattern.CASE_INSENSITIVE | Pattern.DOTALL); + + private final String path; + + private final List<Step> steps; + + private SimplePath(final String path, final List<Step> steps) { + this.path = path; + this.steps = steps; + } + + /** + * Parses a path. + * + * @throws ValidationException if the path is malformed or uses syntax that can point at more + * than one place + */ + public static SimplePath parse(final String path) { + Preconditions.checkNotNull(path, "Path must not be null."); + return new Parser(path).parse(); + } + + /** Returns the steps after {@code $}, so the root path has none. */ + public List<Step> getSteps() { + return steps; + } + + @Override + public String toString() { + return path; + } + + // -------------------------------------------------------------------------------------------- + + /** A step to a field of an object or to an element of an array. */ + @Internal + public static final class Step implements Serializable { + + private static final long serialVersionUID = 1L; + + private final @Nullable String fieldName; + + private final int index; + + private Step(final @Nullable String fieldName, final int index) { + this.fieldName = fieldName; + this.index = index; + } + + static Step field(final String fieldName) { + return new Step(Preconditions.checkNotNull(fieldName), -1); + } + + static Step index(final int index) { + Preconditions.checkArgument(index >= 0, "Index must not be negative."); + return new Step(null, index); + } + + public boolean isField() { + return fieldName != null; + } + + public String getFieldName() { + Preconditions.checkState(fieldName != null, "Step is not a field."); + return fieldName; + } + + public int getIndex() { + Preconditions.checkState(fieldName == null, "Step is not an index."); + return index; + } + + @Override + public boolean equals(final Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final Step that = (Step) o; + return index == that.index && Objects.equals(fieldName, that.fieldName); + } + + @Override + public int hashCode() { + return Objects.hash(fieldName, index); + } + + @Override + public String toString() { + if (fieldName == null) { + return "[" + index + "]"; + } + return "['" + fieldName.replace("\\", "\\\\").replace("'", "\\'") + "']"; + } + } + + // -------------------------------------------------------------------------------------------- + + private static final class Parser { + + private final String path; + + // The end of the path without trailing spaces. + private final int end; + + private int pos; + + private Parser(final String path) { + this.path = path; + int end = path.length(); + while (end > 0 && path.charAt(end - 1) == ' ') { + end--; + } + this.end = end; + } + + private SimplePath parse() { + if (!path.startsWith("$")) { + if (PATH_MODE_PREFIX.matcher(path).matches()) { + throw error("the 'lax' and 'strict' modes are not supported"); + } + throw error("it must start with '$'"); + } + pos = 1; + final List<Step> steps = new ArrayList<>(); + while (pos < end) { + final char c = path.charAt(pos); + if (c == '.') { + pos++; + steps.add(Step.field(readName())); + } else if (c == '[') { + pos++; + steps.add(readBracketStep()); + } else { + throw error(String.format("unexpected '%s' after '%s'", c, consumed())); + } + } + return new SimplePath(path, List.copyOf(steps)); + } + + private String readName() { + final int start = pos; + while (pos < end && isNameCharacter(path.charAt(pos))) { + pos++; + } + if (pos == start) { + throw error(String.format("expected a field name after '%s'", consumed())); + } + return path.substring(start, pos); + } + + private Step readBracketStep() { + final String prefix = path.substring(0, pos - 1); + final char c = pos < end ? path.charAt(pos) : 0; + final Step step; + if (c == '\'' || c == '"') { + pos++; + step = Step.field(readQuotedName(c, prefix)); + } else if (isDigit(c)) { + step = Step.index(readIndex()); + } else { + throw error( + String.format("expected an index or a quoted name after '%s'", consumed())); + } + if (pos >= end || path.charAt(pos) != ']') { + throw error(String.format("expected ']' after '%s'", consumed())); + } + pos++; + return step; + } + + private String readQuotedName(final char quote, final String prefix) { + final StringBuilder name = new StringBuilder(); + while (pos < end) { + final char c = path.charAt(pos); + if (c == quote) { + pos++; + return name.toString(); + } + if (c == '\\') { + final char escaped = pos + 1 < end ? path.charAt(pos + 1) : 0; + if (escaped != '\\' && escaped != '\'' && escaped != '"') { + throw error( + String.format( + "unsupported escape sequence in the name after '%s'. " + + "Only a backslash or a quote can follow a " + + "backslash", + prefix)); + } + name.append(escaped); + pos += 2; + } else { + name.append(c); + pos++; + } + } + throw error(String.format("the name after '%s' has no closing quote", prefix)); + } + + private int readIndex() { + final int start = pos; + while (pos < end && isDigit(path.charAt(pos))) { + pos++; + } + try { + return Integer.parseInt(path.substring(start, pos)); + } catch (NumberFormatException e) { + throw error( + String.format( + "the index after '%s' is larger than %d", + path.substring(0, start), Integer.MAX_VALUE)); + } + } + + private String consumed() { + return path.substring(0, pos); + } + + private ValidationException error(final String reason) { + return new ValidationException( + String.format("Invalid path '%s': %s. %s", path, reason, SYNTAX)); + } + + private static boolean isDigit(final char c) { + return c >= '0' && c <= '9'; + } + + private static boolean isNameCharacter(final char c) { + return NAME_STOP_CHARACTERS.indexOf(c) < 0 + && !Character.isWhitespace(c) + && !Character.isSpaceChar(c) + && !Character.isISOControl(c); + } + } +} diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/SimplePathTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/SimplePathTest.java new file mode 100644 index 00000000000..5a4f7c0beda --- /dev/null +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/SimplePathTest.java @@ -0,0 +1,114 @@ +/* + * 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.utils; + +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.utils.SimplePath.Step; +import org.apache.flink.util.InstantiationUtil; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.List; +import java.util.stream.Stream; + +import static org.apache.flink.table.utils.SimplePath.Step.field; +import static org.apache.flink.table.utils.SimplePath.Step.index; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.params.provider.Arguments.arguments; + +/** Tests for {@link SimplePath}. */ +class SimplePathTest { + + private static Stream<Arguments> validPaths() { + return Stream.of( + arguments("$", List.of()), + arguments("$.items[0].qty", List.of(field("items"), index(0), field("qty"))), + arguments("$.a.b", List.of(field("a"), field("b"))), + arguments("$[007]", List.of(index(7))), + arguments("$[2147483647]", List.of(index(Integer.MAX_VALUE))), + arguments("$[\"a.b\"]", List.of(field("a.b"))), + arguments("$['it\\'s']", List.of(field("it's"))), + // The padding of a CHAR(n) value. + arguments("$.a ", List.of(field("a")))); + } + + @ParameterizedTest(name = "[{index}] {0}") + @MethodSource("validPaths") + void testParse(final String path, final List<Step> steps) { + assertThat(SimplePath.parse(path).getSteps()).isEqualTo(steps); + } + + @ParameterizedTest(name = "[{index}] {0}") + @CsvSource( + delimiter = '|', + quoteCharacter = '`', + value = { + "` $.a` | it must start with '$'", + "lax $.a | the 'lax' and 'strict' modes are not supported", + "$.* | expected a field name after '$.'", + "$[-1] | expected an index or a quoted name after '$['", + "$[0,1] | expected ']' after '$[0'", + "$['a\\'] | the name after '$' has no closing quote", + "$.x['a\\nb'] | unsupported escape sequence in the name after '$.x'", + "$[2147483648] | the index after '$[' is larger than 2147483647" + }) + void testParseFails(final String path, final String reason) { + assertThatThrownBy(() -> SimplePath.parse(path)) + .isInstanceOf(ValidationException.class) + .hasMessageStartingWith(String.format("Invalid path '%s': %s.", path, reason)); + } + + @ParameterizedTest(name = "[{index}] U+{0}") + @ValueSource(chars = {']', '(', ')', '*', '\'', '"', '\\', ' ', '\t', '\u00A0', '\u0001'}) + void testFieldNameStopsAt(final char c) { + final String path = "$.a" + c + "b"; + assertThatThrownBy(() -> SimplePath.parse(path)) + .isInstanceOf(ValidationException.class) + .hasMessageStartingWith( + String.format("Invalid path '%s': unexpected '%s' after '$.a'.", path, c)); + } + + @Test + void testErrorMessageExplainsTheSyntax() { + assertThatThrownBy(() -> SimplePath.parse("$.a b")) + .hasMessage( + "Invalid path '$.a b': unexpected ' ' after '$.a'. A path is '$' followed " + + "by the steps .name, ['name'], [\"name\"] or [index], without " + + "wildcards, filters, slices, negative indices or functions. " + + "Use brackets for a name that has whitespace or one of the " + + "characters . [ ] ( ) * ' \" \\ in it."); + } + + @Test + void testSerializable() throws Exception { + final SimplePath path = SimplePath.parse("$.items[0]['sku']"); + + final SimplePath copy = + InstantiationUtil.clone(path, SimplePathTest.class.getClassLoader()); + + assertThat(copy.getSteps()).isEqualTo(path.getSteps()); + assertThat(copy).hasToString("$.items[0]['sku']"); + } +} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java index 5163074334e..ea1233c9b47 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java @@ -22,7 +22,7 @@ import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.config.AggregatePhaseStrategy; import org.apache.flink.table.api.config.ExecutionConfigOptions; -import org.apache.flink.table.api.config.TableConfigOptions; +import org.apache.flink.table.api.internal.TableConfigValidation; import org.apache.flink.table.planner.calcite.CalciteConfig; import org.apache.flink.table.planner.calcite.CalciteConfig$; import org.apache.flink.table.planner.plan.utils.OperatorType; @@ -33,7 +33,6 @@ import java.util.Set; import static org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS; import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY; -import static org.apache.flink.table.api.internal.TableConfigValidation.validateTimeZone; /** Utility class for {@link TableConfig} related helper functions. */ public class TableConfigUtils { @@ -97,12 +96,7 @@ public class TableConfigUtils { * @see TableConfig#getLocalTimeZone() */ public static ZoneId getLocalTimeZone(ReadableConfig tableConfig) { - final String zone = tableConfig.get(TableConfigOptions.LOCAL_TIME_ZONE); - if (TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)) { - return ZoneId.systemDefault(); - } - validateTimeZone(zone); - return ZoneId.of(zone); + return TableConfigValidation.getLocalTimeZone(tableConfig); } @Deprecated diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java new file mode 100644 index 00000000000..6780b01b5cc --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java @@ -0,0 +1,167 @@ +/* + * 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.functions; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.api.config.TableConfigOptions; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; + +import java.time.Instant; +import java.util.stream.Stream; + +import static org.apache.flink.table.api.DataTypes.DECIMAL; +import static org.apache.flink.table.api.DataTypes.INT; +import static org.apache.flink.table.api.DataTypes.STRING; +import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ; +import static org.apache.flink.table.api.DataTypes.VARIANT; +import static org.apache.flink.table.api.Expressions.$; +import static org.apache.flink.table.api.Expressions.jsonString; +import static org.apache.flink.table.api.Expressions.lit; +import static org.apache.flink.table.api.Expressions.nullOf; + +/** + * Tests for the VARIANT functions that are not covered by the JSON and XML function tests. + * + * <p>Path syntax is covered by {@code SimplePathTest}. A path that changes from row to row is + * covered by {@code VariantSemanticTest}. + */ +class VariantFunctionsITCase extends BuiltInFunctionTestBase { + + private static final String ORDER = + "{\"order_id\": 17, \"user\": {\"name\": \"ada\", \"email\": \"[email protected]\"}, " + + "\"items\": [{\"sku\": \"A1\", \"qty\": 2}], \"coupon\": null, " + + "\"price\": \"12.50\"}"; + + @Override + Configuration getConfiguration() { + return super.getConfiguration().set(TableConfigOptions.LOCAL_TIME_ZONE, "Asia/Shanghai"); + } + + @Override + Stream<TestSetSpec> getTestSetSpecs() { + return Stream.of(variantGet(), variantGetFailure(), tryVariantGet()); + } + + private static TestSetSpec variantGet() { + return TestSetSpec.forFunction(BuiltInFunctionDefinitions.VARIANT_GET) + .onFieldsWithData(ORDER, "$.user.name", null, Instant.EPOCH) + .andDataTypes(STRING().notNull(), STRING().notNull(), STRING(), TIMESTAMP_LTZ(3)) + .testResult( + $("f0").parseJson().variantGet("$.order_id", INT()), + "VARIANT_GET(PARSE_JSON(f0), '$.order_id', 'int')", + 17, + INT()) + .testResult( + jsonString($("f0").parseJson().variantGet("$.user.name", VARIANT())), + "JSON_STRING(VARIANT_GET(PARSE_JSON(f0), '$.user.name', 'VARIANT'))", + "\"ada\"", + STRING()) + // A path that points at nothing gives NULL, so even a NOT NULL type is nullable. + .testResult( + $("f0").parseJson().variantGet("$.missing", INT().notNull()), + "VARIANT_GET(PARSE_JSON(f0), '$.missing', 'INT NOT NULL')", + null, + INT()) + // A variant null stays a VARIANT, and casts to NULL. + .testResult( + jsonString($("f0").parseJson().variantGet("$.coupon")), + "JSON_STRING(VARIANT_GET(PARSE_JSON(f0), '$.coupon'))", + "null", + STRING()) + .testResult( + $("f0").parseJson().variantGet("$.coupon", STRING()), + "VARIANT_GET(PARSE_JSON(f0), '$.coupon', 'STRING')", + null, + STRING()) + .testResult( + $("f0").parseJson().variantGet($("f1"), STRING()), + "VARIANT_GET(PARSE_JSON(f0), f1, 'STRING')", + "ada", + STRING()) + .testResult( + $("f2").parseJson().variantGet("$.order_id", INT()), + "VARIANT_GET(PARSE_JSON(f2), '$.order_id', 'INT')", + null, + INT()) + .testResult( + $("f0").parseJson().variantGet(nullOf(STRING())), + "VARIANT_GET(PARSE_JSON(f0), NULL)", + null, + VARIANT()) + // A TIMESTAMP_LTZ renders in the session time zone, as in CAST. + .testResult( + $("f3").cast(VARIANT()).variantGet("$", STRING()), + "VARIANT_GET(CAST(f3 AS VARIANT), '$', 'STRING')", + "1970-01-01 08:00:00.000000", + STRING()); + } + + private static TestSetSpec variantGetFailure() { + return TestSetSpec.forFunction(BuiltInFunctionDefinitions.VARIANT_GET, "failures") + .onFieldsWithData(ORDER) + .andDataTypes(STRING().notNull()) + .testSqlRuntimeError( + "VARIANT_GET(PARSE_JSON(f0), '$.price', 'DECIMAL(4, 2)')", + TableRuntimeException.class, + "Failed to cast the VARIANT value at path '$.price' to DECIMAL(4, 2): " + + "Cannot cast a VARIANT STRING value to DECIMAL(4, 2).") + // A path that becomes a literal only during optimization fails at run time. + .testTableApiRuntimeError( + $("f0").parseJson().variantGet(lit("$.user[").cast(STRING())), + TableRuntimeException.class, + "Invalid path '$.user['") + .testSqlValidationError( + "VARIANT_GET(PARSE_JSON(f0), '$.items[-1]')", + "Invalid path '$.items[-1]': expected an index or a quoted name after " + + "'$.items['.") + .testSqlValidationError( + "VARIANT_GET(PARSE_JSON(f0), '$.a', 'MULTISET<INT>')", + "VARIANT_GET cannot return type MULTISET<INT>, because a VARIANT cannot " + + "be cast to it.") + .testSqlValidationError( + "VARIANT_GET(PARSE_JSON(f0), '$.a', 'NOT A TYPE')", + "Invalid type 'NOT A TYPE' for VARIANT_GET:") + .testSqlValidationError( + "VARIANT_GET(PARSE_JSON(f0), '$.a', f0)", + "Invalid input arguments. Expected signatures are:"); + } + + private static TestSetSpec tryVariantGet() { + return TestSetSpec.forFunction(BuiltInFunctionDefinitions.TRY_VARIANT_GET) + .onFieldsWithData(ORDER) + .andDataTypes(STRING().notNull()) + .testResult( + $("f0").parseJson().tryVariantGet("$.price", DECIMAL(4, 2)), + "TRY_VARIANT_GET(PARSE_JSON(f0), '$.price', 'DECIMAL(4, 2)')", + null, + DECIMAL(4, 2)) + // A path that becomes a literal only during optimization gives NULL. + .testResult( + $("f0").parseJson().tryVariantGet(lit("$.user[").cast(STRING()), STRING()), + "TRY_VARIANT_GET(PARSE_JSON(f0), CAST('$.user[' AS STRING), 'STRING')", + null, + STRING()) + // A malformed literal path still fails at planning time. + .testTableApiValidationError( + $("f0").parseJson().tryVariantGet("$.user["), + "Invalid path '$.user[': expected an index or a quoted name after " + + "'$.user['."); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java index 580201100f4..1fba4b01203 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java @@ -112,6 +112,56 @@ public class VariantSemanticTest extends SemanticTestBase { .runSql("INSERT INTO sink_t SELECT TRY_PARSE_JSON(v) FROM t") .build(); + static final String JSON_OBJECT_A_AND_B = "{\"a\": 1, \"b\": 2}"; + + static final TableTestProgram TRY_VARIANT_GET_PATH_PER_ROW = + TableTestProgram.of( + "try-variant-get-path-per-row", + "validates TRY_VARIANT_GET reads a path that changes from row to row") + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema("v STRING", "p STRING") + .producedValues( + Row.of(JSON_OBJECT_A_AND_B, "$.a"), + Row.of(JSON_OBJECT_A_AND_B, "$.c"), + Row.of(JSON_OBJECT_A_AND_B, "$.b"), + Row.of(JSON_OBJECT_A_AND_B, "$.a["), + // The path before the malformed one again, which a + // cache that keeps its key on an error gets wrong. + Row.of(JSON_OBJECT_A_AND_B, "$.b")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema("i INT") + .consumedValues( + Row.of(1), new Row(1), Row.of(2), new Row(1), Row.of(2)) + .build()) + .runSql( + "INSERT INTO sink_t SELECT TRY_VARIANT_GET(PARSE_JSON(v), p, 'INT') FROM t") + .build(); + + static final TableTestProgram VARIANT_GET_FAIL_MALFORMED_PATH = + TableTestProgram.of( + "variant-get-fail-malformed-path", + "validates VARIANT_GET throws on a malformed path that is not a literal") + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema("v STRING", "p STRING") + .producedValues( + Row.of(JSON_OBJECT_A_AND_B, "$.a"), + Row.of(JSON_OBJECT_A_AND_B, "$.a[")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema("i INT") + .consumedValues(Row.of(1)) + .build()) + .runFailingSql( + "INSERT INTO sink_t SELECT VARIANT_GET(PARSE_JSON(v), p, 'INT') FROM t", + TableRuntimeException.class, + "Invalid path '$.a[': expected an index or a quoted name after '$.a['.") + .build(); + static final TableTestProgram VARIANT_OMITTED_FROM_INSERT_COLUMN_LIST = TableTestProgram.of( "variant-omitted-from-insert-column-list", @@ -683,6 +733,8 @@ public class VariantSemanticTest extends SemanticTestBase { PARSE_JSON_OF_NULL, PARSE_JSON_FAIL_MALFORMED_JSON, TRY_PARSE_JSON_HANDLE_MALFORMED_JSON, + TRY_VARIANT_GET_PATH_PER_ROW, + VARIANT_GET_FAIL_MALFORMED_PATH, BUILTIN_AGG, BUILTIN_AGG_WITH_RETRACTION, VARIANT_AS_UDF_ARG, diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryVariantGetFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryVariantGetFunction.java new file mode 100644 index 00000000000..b6f0d36a71e --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryVariantGetFunction.java @@ -0,0 +1,36 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext; + +/** + * Implementation of {@link BuiltInFunctionDefinitions#TRY_VARIANT_GET}, which returns {@code NULL} + * where {@link VariantGetFunction} fails: for a malformed path that is not a literal, or a value + * that cannot be cast to the result type. + */ +@Internal +public class TryVariantGetFunction extends VariantGetFunction { + + public TryVariantGetFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.TRY_VARIANT_GET, context, true); + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java new file mode 100644 index 00000000000..d6d1facbb45 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java @@ -0,0 +1,182 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableConfigValidation; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.utils.FromVariantConverter; +import org.apache.flink.table.functions.BuiltInFunctionDefinition; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext; +import org.apache.flink.table.types.inference.CallContext; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.utils.SimplePath; +import org.apache.flink.types.variant.Variant; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Optional; + +/** + * Implementation of {@link BuiltInFunctionDefinitions#VARIANT_GET}. + * + * <p>A path that points at nothing gives {@code NULL}: a missing field, an index past the end, a + * field of a non-object or an element of a non-array. The value found is cast to the result type + * the way {@code CAST} does, so a variant null becomes {@code NULL} unless the result is a {@code + * VARIANT}. + */ +@Internal +public class VariantGetFunction extends BuiltInScalarFunction { + + private static final int ARG_PATH = 1; + + private final boolean isTry; + + private final @Nullable SimplePath literalPath; + + // Null when the result is a VARIANT, which needs no cast. + private final @Nullable FromVariantConverter converter; + + private final String targetTypeName; + + private transient @Nullable String lastPathString; + + private transient @Nullable SimplePath lastPath; + + private transient @Nullable ValidationException lastPathError; + + public VariantGetFunction(SpecializedContext context) { + this(BuiltInFunctionDefinitions.VARIANT_GET, context, false); + } + + VariantGetFunction( + BuiltInFunctionDefinition definition, SpecializedContext context, boolean isTry) { + super(definition, context); + this.isTry = isTry; + this.literalPath = parseLiteralPath(context.getCallContext()); + final LogicalType outputType = getOutputDataType().getLogicalType(); + this.converter = + outputType.is(LogicalTypeRoot.VARIANT) + ? null + : FromVariantConverter.create( + outputType, + TableConfigValidation.getLocalTimeZone(context.getConfiguration())); + this.targetTypeName = outputType.asSummaryString(); + } + + public @Nullable Object eval(@Nullable Variant variant, @Nullable StringData path) { + return eval(variant, path, null); + } + + public @Nullable Object eval( + @Nullable Variant variant, @Nullable StringData path, @Nullable StringData type) { + // The type only matters for planning, where it defines the output type. + if (variant == null || path == null) { + return null; + } + final SimplePath simplePath = literalPath != null ? literalPath : parsePath(path); + if (simplePath == null) { + return null; + } + final Variant value = navigate(variant, simplePath); + if (value == null || converter == null) { + return value; + } + if (value.isNull()) { + return null; + } + try { + return converter.convert(value); + } catch (RuntimeException e) { + if (isTry) { + return null; + } + throw new TableRuntimeException( + String.format( + "Failed to cast the VARIANT value at path '%s' to %s: %s", + simplePath, targetTypeName, e.getMessage()), + e); + } + } + + /** + * Returns the literal path, or {@code null} to parse the path of each row. + * + * <p>The input strategy rejects a malformed literal path. A path that becomes a literal only + * during optimization, such as a constant column of a view, skips that check, so a malformed + * one is left to the rules for a path that is not a literal. + */ + private static @Nullable SimplePath parseLiteralPath(CallContext callContext) { + if (!callContext.isArgumentLiteral(ARG_PATH)) { + return null; + } + final Optional<String> path = callContext.getArgumentValue(ARG_PATH, String.class); + if (path.isEmpty()) { + return null; + } + try { + return SimplePath.parse(path.get()); + } catch (ValidationException e) { + return null; + } + } + + /** Returns the path, or {@code null} for a malformed path in {@code TRY_VARIANT_GET}. */ + private @Nullable SimplePath parsePath(StringData path) { + final String pathString = path.toString(); + if (!pathString.equals(lastPathString)) { + lastPathString = pathString; + try { + lastPath = SimplePath.parse(pathString); + lastPathError = null; + } catch (ValidationException e) { + lastPath = null; + lastPathError = e; + } + } + if (lastPathError != null && !isTry) { + throw new TableRuntimeException(lastPathError.getMessage(), lastPathError); + } + return lastPath; + } + + /** Returns the value at the path, or {@code null} if the path points at nothing. */ + @VisibleForTesting + static @Nullable Variant navigate(Variant variant, SimplePath path) { + final List<SimplePath.Step> steps = path.getSteps(); + Variant current = variant; + for (final SimplePath.Step step : steps) { + if (step.isField()) { + current = current.isObject() ? current.getField(step.getFieldName()) : null; + } else { + current = current.isArray() ? current.getElement(step.getIndex()) : null; + } + if (current == null) { + return null; + } + } + return current; + } +}
