This is an automated email from the ASF dual-hosted git repository. AHeise pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit e785c495e43c5e5b1357c8f829204a3272d1cd70 Author: Ramin Gharib <[email protected]> AuthorDate: Mon Sep 28 11:26:36 2026 +0200 [FLINK-40825][table] Support casting primitive types to VARIANT CAST and TRY_CAST now convert BOOLEAN, numeric, character string, binary string, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, and UUID values to VARIANT. SqlCastFunction rejected every cast to VARIANT, and now routes a VARIANT target to LogicalTypeCasts, which lists the supported sources explicitly. A type without a VARIANT kind, such as INTERVAL, is rejected at validation, and the cast is explicit only. The value keeps the kind of its SQL type, so an integer keeps its width and a string is [...] A timestamp keeps its declared precision: up to 6 it is stored with microseconds and above with nanoseconds. Nanoseconds only cover 1677-09-21 to 2262-04-11, so such a value outside that range fails the cast. A VARIANT holds at most 16 MiB, so a longer string or binary value fails the cast too. The rule reports a type as fallible when its precision or declared length allows such a value, so TRY_CAST returns NULL for it. A constructed cast checks each child with the same rules, so a cast such as ARRAY<INT> to ARRAY<VARIANT> works element by element. Casting a whole ARRAY, MAP, or ROW into one VARIANT follows in FLINK-40826. --- docs/content.zh/docs/sql/reference/data-types.md | 87 +++++-- docs/content/docs/sql/reference/data-types.md | 87 +++++-- .../types/logical/utils/LogicalTypeCasts.java | 27 ++ .../flink/table/types/LogicalTypeCastsTest.java | 69 ++++- .../apache/calcite/sql/fun/SqlCastFunction.java | 14 +- .../functions/casting/CastRuleProvider.java | 1 + .../casting/PrimitiveToVariantCastRule.java | 179 +++++++++++++ .../planner/functions/CastFunctionITCase.java | 277 +++++++++++++++++++++ .../functions/casting/CastRuleProviderTest.java | 73 ++++++ .../planner/functions/casting/CastRulesTest.java | 176 +++++++++++++ .../table/runtime/functions/VariantCastUtils.java | 147 ++++++++++- .../runtime/functions/VariantCastUtilsTest.java | 27 ++ 12 files changed, 1116 insertions(+), 48 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 4858d42d889..e5344621c3f 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1517,8 +1517,9 @@ particularly useful in dynamic environments where schemas may evolve over time. A `VARIANT` stores a single value of one of the following kinds: `NULL`, `BOOLEAN`, `TINYINT`, `SMALLINT`, `INT`, `BIGINT`, `FLOAT`, `DOUBLE`, `DECIMAL` (up to precision 38), `STRING`, `DATE`, -`TIMESTAMP`, `TIMESTAMP_LTZ`, `BYTES`, or a nested array or object. `TIMESTAMP` and `TIMESTAMP_LTZ` -are stored with microsecond precision. +`TIME`, `TIMESTAMP`, `TIMESTAMP_LTZ`, `BYTES`, `UUID`, or a nested array or object. `TIME` is stored +with microsecond precision, and `TIMESTAMP` and `TIMESTAMP_LTZ` with microsecond or nanosecond +precision. The `PARSE_JSON` function produces only the kinds that JSON syntax can express: @@ -1542,8 +1543,8 @@ The JSON specification has no `NaN` or infinity literals, so `PARSE_JSON('NaN')` `PARSE_JSON('Infinity')`, and `PARSE_JSON('-Infinity')` fail. `PARSE_JSON('1e400')` fails as well because a value outside the `DOUBLE` range cannot be stored as a finite number. In all of these cases `TRY_PARSE_JSON` returns `NULL`. -A `VARIANT` has no dedicated kind for these values. To keep one, store it as a JSON string and cast -it back out, for example `CAST(CAST(PARSE_JSON('"Infinity"') AS STRING) AS FLOAT)`. +To store one of these values, cast the number to a `VARIANT` instead of parsing it, for example +`CAST(CAST('Infinity' AS DOUBLE) AS VARIANT)`. 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 @@ -1669,6 +1670,56 @@ CAST(o AS MAP<STRING, VARIANT>) -- values kept as variants, the variant null in CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a character string ``` +A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a type that a +`VARIANT` kind holds without loss is supported, and any other type, such as `INTERVAL`, `RAW`, or +`BITMAP`, is rejected at validation. The value keeps the kind of its SQL type: + +| Input type | Stored `VARIANT` kind | +|--------------------------------------------|-------------------------------------------------| +| `BOOLEAN` | `BOOLEAN` | +| `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT` | `TINYINT`, `SMALLINT`, `INT`, `BIGINT` | +| `FLOAT`, `DOUBLE`, `DECIMAL` | `FLOAT`, `DOUBLE`, `DECIMAL` | +| `CHAR`, `VARCHAR`, `STRING` | `STRING` | +| `BINARY`, `VARBINARY`, `BYTES` | `BYTES` | +| `DATE`, `TIME`, `UUID` | `DATE`, `TIME`, `UUID` | +| `TIMESTAMP`, `TIMESTAMP_LTZ` | `TIMESTAMP`, `TIMESTAMP_LTZ` | + +- An integer keeps the width of its SQL type, so a `BIGINT` is stored as a `BIGINT` even when the + value would fit a smaller kind. `PARSE_JSON('1')` instead picks the smallest kind, a `TINYINT`. + Either way it casts back to any integer type that holds the value. +- A character string is stored as a `STRING` and is never parsed. Use `PARSE_JSON` to parse JSON text. +- A `TIMESTAMP(p)` or `TIMESTAMP_LTZ(p)` keeps its declared precision. Up to a precision of 6 it is + stored with microseconds, and above with nanoseconds, even when the value has no digits below a + microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value + outside that range fails the cast, and `TRY_CAST` returns `NULL`. +- A `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although `PARSE_JSON` rejects them. A + `VARIANT` is not limited to JSON, but JSON has no literal for them. So `JSON_STRING` fails on such + a variant, and so does writing it to a sink with the `json` or `raw` format, which fails the job. + Printing it and casting it to `STRING` show `NaN`, and it casts back to `FLOAT` or `DOUBLE` + unchanged. +- A `VARIANT` holds at most 16 MiB, so a string or binary value can have at most 16,777,211 bytes. + A longer value fails the cast, and `TRY_CAST` returns `NULL`. +- A SQL `NULL` casts to a SQL `NULL`, not to a variant null. +- The cast is never applied implicitly, so writing an `INT` into a `VARIANT` column needs an explicit + `CAST`. +- A constructed type casts element by element when its element, field, or value type casts to + `VARIANT`, for example `ARRAY<INT>` to `ARRAY<VARIANT>`. A `NULL` element stays a SQL `NULL`. A + whole `ARRAY`, `MAP`, or `ROW` does not cast into a single `VARIANT` yet. +- Two `VARIANT` values are equal only when their binary encodings match. So as a `MAP` key or a + `MULTISET` element, a `1` cast from `INT` does not match a `1` cast from `BIGINT` or parsed by + `PARSE_JSON('1')`. + +```sql +CAST(42 AS VARIANT) -- 42, stored as an INT +CAST(CAST(1 AS BIGINT) AS VARIANT) -- 1, stored as a BIGINT +CAST('{"a": 1}' AS VARIANT) -- the string '{"a": 1}', not an object +CAST(NULL AS VARIANT) -- NULL +CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE +CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation +CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1, NULL], each element a VARIANT, the NULL stays SQL NULL +CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not supported yet +``` + **Declaration** {{< tabs "25c30432-8460-441d-a036-9416d8202882" >}} @@ -1865,20 +1916,20 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 | Input\Target | `CHAR`¹/<br/>`VARCHAR`¹/<br/>`STRING` | `BINARY`¹/<br/>`VARBINARY`¹/<br/>`BYTES` | `BOOLEAN` | `DECIMAL` | `TINYINT` | `SMALLINT` | `INTEGER` | `BIGINT` | `FLOAT` | `DOUBLE` | `DATE` | `TIME` | `TIMESTAMP` | `TIMESTAMP_LTZ` | `INTERVAL` | `ARRAY` | `MULTISET` | `MAP` | `ROW` | `STRUCTURED` | `RAW` | `VARIANT` | `BITMAP` | |:---------------------------------------|:-------------------------------------:|:----------------------------------------:|:---------:|:---------:|:---------:|:----------:|:---------:|:--------:|:-------:|:--------:|:------:|:------:|:-----------:|:---------------:|:----------:|:-------:|:----------:|:-----:|:-----:|:------------:|:-----:|:---------:|:--------:| -| `CHAR`/<br/>`VARCHAR`/<br/>`STRING` | Y | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | N | N | N | N | N | N | N | N | -| `BINARY`/<br/>`VARBINARY`/<br/>`BYTES` | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `BOOLEAN` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DECIMAL` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `TINYINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | N | N | -| `SMALLINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | N | N | -| `INTEGER` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁵ | N | N | N | N | N | N | N | N | -| `BIGINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁶ | N | N | N | N | N | N | N | N | -| `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | +| `CHAR`/<br/>`VARCHAR`/<br/>`STRING` | Y | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | N | N | N | N | N | N | ! | N | +| `BINARY`/<br/>`VARBINARY`/<br/>`BYTES` | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | ! | N | +| `BOOLEAN` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DECIMAL` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `TINYINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | Y | N | +| `SMALLINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | Y | N | +| `INTEGER` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁵ | N | N | N | N | N | N | Y | N | +| `BIGINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁶ | N | N | N | N | N | N | Y | N | +| `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | +| `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `INTERVAL` | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | N | N | N | N | | `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | N | | `MULTISET` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index a3b4371b4b1..ca35e7dac3f 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1525,8 +1525,9 @@ particularly useful in dynamic environments where schemas may evolve over time. A `VARIANT` stores a single value of one of the following kinds: `NULL`, `BOOLEAN`, `TINYINT`, `SMALLINT`, `INT`, `BIGINT`, `FLOAT`, `DOUBLE`, `DECIMAL` (up to precision 38), `STRING`, `DATE`, -`TIMESTAMP`, `TIMESTAMP_LTZ`, `BYTES`, or a nested array or object. `TIMESTAMP` and `TIMESTAMP_LTZ` -are stored with microsecond precision. +`TIME`, `TIMESTAMP`, `TIMESTAMP_LTZ`, `BYTES`, `UUID`, or a nested array or object. `TIME` is stored +with microsecond precision, and `TIMESTAMP` and `TIMESTAMP_LTZ` with microsecond or nanosecond +precision. The `PARSE_JSON` function produces only the kinds that JSON syntax can express: @@ -1550,8 +1551,8 @@ The JSON specification has no `NaN` or infinity literals, so `PARSE_JSON('NaN')` `PARSE_JSON('Infinity')`, and `PARSE_JSON('-Infinity')` fail. `PARSE_JSON('1e400')` fails as well because a value outside the `DOUBLE` range cannot be stored as a finite number. In all of these cases `TRY_PARSE_JSON` returns `NULL`. -A `VARIANT` has no dedicated kind for these values. To keep one, store it as a JSON string and cast -it back out, for example `CAST(CAST(PARSE_JSON('"Infinity"') AS STRING) AS FLOAT)`. +To store one of these values, cast the number to a `VARIANT` instead of parsing it, for example +`CAST(CAST('Infinity' AS DOUBLE) AS VARIANT)`. 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 @@ -1677,6 +1678,56 @@ CAST(o AS MAP<STRING, VARIANT>) -- values kept as variants, the variant null in CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a character string ``` +A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a type that a +`VARIANT` kind holds without loss is supported, and any other type, such as `INTERVAL`, `RAW`, or +`BITMAP`, is rejected at validation. The value keeps the kind of its SQL type: + +| Input type | Stored `VARIANT` kind | +|--------------------------------------------|-------------------------------------------------| +| `BOOLEAN` | `BOOLEAN` | +| `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT` | `TINYINT`, `SMALLINT`, `INT`, `BIGINT` | +| `FLOAT`, `DOUBLE`, `DECIMAL` | `FLOAT`, `DOUBLE`, `DECIMAL` | +| `CHAR`, `VARCHAR`, `STRING` | `STRING` | +| `BINARY`, `VARBINARY`, `BYTES` | `BYTES` | +| `DATE`, `TIME`, `UUID` | `DATE`, `TIME`, `UUID` | +| `TIMESTAMP`, `TIMESTAMP_LTZ` | `TIMESTAMP`, `TIMESTAMP_LTZ` | + +- An integer keeps the width of its SQL type, so a `BIGINT` is stored as a `BIGINT` even when the + value would fit a smaller kind. `PARSE_JSON('1')` instead picks the smallest kind, a `TINYINT`. + Either way it casts back to any integer type that holds the value. +- A character string is stored as a `STRING` and is never parsed. Use `PARSE_JSON` to parse JSON text. +- A `TIMESTAMP(p)` or `TIMESTAMP_LTZ(p)` keeps its declared precision. Up to a precision of 6 it is + stored with microseconds, and above with nanoseconds, even when the value has no digits below a + microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value + outside that range fails the cast, and `TRY_CAST` returns `NULL`. +- A `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although `PARSE_JSON` rejects them. A + `VARIANT` is not limited to JSON, but JSON has no literal for them. So `JSON_STRING` fails on such + a variant, and so does writing it to a sink with the `json` or `raw` format, which fails the job. + Printing it and casting it to `STRING` show `NaN`, and it casts back to `FLOAT` or `DOUBLE` + unchanged. +- A `VARIANT` holds at most 16 MiB, so a string or binary value can have at most 16,777,211 bytes. + A longer value fails the cast, and `TRY_CAST` returns `NULL`. +- A SQL `NULL` casts to a SQL `NULL`, not to a variant null. +- The cast is never applied implicitly, so writing an `INT` into a `VARIANT` column needs an explicit + `CAST`. +- A constructed type casts element by element when its element, field, or value type casts to + `VARIANT`, for example `ARRAY<INT>` to `ARRAY<VARIANT>`. A `NULL` element stays a SQL `NULL`. A + whole `ARRAY`, `MAP`, or `ROW` does not cast into a single `VARIANT` yet. +- Two `VARIANT` values are equal only when their binary encodings match. So as a `MAP` key or a + `MULTISET` element, a `1` cast from `INT` does not match a `1` cast from `BIGINT` or parsed by + `PARSE_JSON('1')`. + +```sql +CAST(42 AS VARIANT) -- 42, stored as an INT +CAST(CAST(1 AS BIGINT) AS VARIANT) -- 1, stored as a BIGINT +CAST('{"a": 1}' AS VARIANT) -- the string '{"a": 1}', not an object +CAST(NULL AS VARIANT) -- NULL +CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE +CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation +CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1, NULL], each element a VARIANT, the NULL stays SQL NULL +CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not supported yet +``` + **Declaration** {{< tabs "25c30432-8460-441d-a036-9416d8202882" >}} @@ -1874,20 +1925,20 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | Input\Target | `CHAR`¹/<br/>`VARCHAR`¹/<br/>`STRING` | `BINARY`¹/<br/>`VARBINARY`¹/<br/>`BYTES` | `BOOLEAN` | `DECIMAL` | `TINYINT` | `SMALLINT` | `INTEGER` | `BIGINT` | `FLOAT` | `DOUBLE` | `DATE` | `TIME` | `TIMESTAMP` | `TIMESTAMP_LTZ` | `INTERVAL` | `ARRAY` | `MULTISET` | `MAP` | `ROW` | `STRUCTURED` | `RAW` | `VARIANT` | `BITMAP` | |:---------------------------------------|:-------------------------------------:|:----------------------------------------:|:---------:|:---------:|:---------:|:----------:|:---------:|:--------:|:-------:|:--------:|:------:|:------:|:-----------:|:---------------:|:----------:|:-------:|:----------:|:-----:|:-----:|:------------:|:-----:|:---------:|:--------:| -| `CHAR`/<br/>`VARCHAR`/<br/>`STRING` | Y | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | N | N | N | N | N | N | N | N | -| `BINARY`/<br/>`VARBINARY`/<br/>`BYTES` | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `BOOLEAN` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DECIMAL` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `TINYINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | N | N | -| `SMALLINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | N | N | -| `INTEGER` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁵ | N | N | N | N | N | N | N | N | -| `BIGINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁶ | N | N | N | N | N | N | N | N | -| `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | -| `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | -| `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | +| `CHAR`/<br/>`VARCHAR`/<br/>`STRING` | Y | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | N | N | N | N | N | N | ! | N | +| `BINARY`/<br/>`VARBINARY`/<br/>`BYTES` | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | ! | N | +| `BOOLEAN` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DECIMAL` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `TINYINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | Y | N | +| `SMALLINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | N | N | Y | N | +| `INTEGER` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁵ | N | N | N | N | N | N | Y | N | +| `BIGINT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁶ | N | N | N | N | N | N | Y | N | +| `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | +| `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | +| `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `INTERVAL` | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | N | N | N | N | | `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | N | | `MULTISET` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java index d351a36746d..ab0036b5a70 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java @@ -403,6 +403,33 @@ public final class LogicalTypeCasts { .explicitFromFamily(EXACT_NUMERIC, CHARACTER_STRING) .build(); + // ----------------------------------------------------------------------------------------- + // VARIANT type + // ----------------------------------------------------------------------------------------- + + // Only a type with a VARIANT kind that holds its value without loss casts to VARIANT. + castTo(VARIANT) + .implicitFrom(VARIANT) + .explicitFrom( + BOOLEAN, + TINYINT, + SMALLINT, + INTEGER, + BIGINT, + FLOAT, + DOUBLE, + DECIMAL, + CHAR, + VARCHAR, + BINARY, + VARBINARY, + DATE, + TIME_WITHOUT_TIME_ZONE, + TIMESTAMP_WITHOUT_TIME_ZONE, + TIMESTAMP_WITH_LOCAL_TIME_ZONE, + UUID) + .build(); + // ----------------------------------------------------------------------------------------- // UUID type // ----------------------------------------------------------------------------------------- diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java index 1e96b65a130..a238417654b 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java @@ -390,8 +390,73 @@ class LogicalTypeCastsTest { // numeric is not castable to or from UUID Arguments.of(new UuidType(), new IntType(), false, false), Arguments.of(new IntType(), new UuidType(), false, false), - // a UUID does not cast to VARIANT - Arguments.of(new UuidType(), new VariantType(), false, false)); + // a type with a VARIANT kind casts to VARIANT, explicitly only + Arguments.of(new BooleanType(), new VariantType(), false, true), + Arguments.of(new TinyIntType(), new VariantType(), false, true), + Arguments.of(new SmallIntType(), new VariantType(), false, true), + Arguments.of(new IntType(), new VariantType(), false, true), + Arguments.of(new BigIntType(), new VariantType(), false, true), + Arguments.of(new FloatType(), new VariantType(), false, true), + Arguments.of(new DoubleType(), new VariantType(), false, true), + Arguments.of(new DecimalType(38, 10), new VariantType(), false, true), + Arguments.of(new CharType(5), new VariantType(), false, true), + Arguments.of(VarCharType.STRING_TYPE, new VariantType(), false, true), + Arguments.of(new BinaryType(4), new VariantType(), false, true), + Arguments.of( + new VarBinaryType(VarBinaryType.MAX_LENGTH), + new VariantType(), + false, + true), + Arguments.of(new DateType(), new VariantType(), false, true), + Arguments.of(new TimeType(), new VariantType(), false, true), + Arguments.of(new TimestampType(9), new VariantType(), false, true), + Arguments.of(new LocalZonedTimestampType(3), new VariantType(), false, true), + Arguments.of(new UuidType(), new VariantType(), false, true), + Arguments.of(new NullType(), new VariantType(), true, true), + // a type without a VARIANT kind does not cast to VARIANT + Arguments.of( + new YearMonthIntervalType(YearMonthIntervalType.YearMonthResolution.MONTH), + new VariantType(), + false, + false), + Arguments.of(new ZonedTimestampType(), new VariantType(), false, false), + Arguments.of( + new MultisetType(VarCharType.STRING_TYPE), new VariantType(), false, false), + // a constructed type casts element by element when each child casts to VARIANT + Arguments.of( + new ArrayType(new IntType()), + new ArrayType(new VariantType()), + false, + true), + Arguments.of( + new RowType(List.of(new RowField("a", new IntType()))), + new RowType(List.of(new RowField("a", new VariantType()))), + false, + true), + Arguments.of( + new MapType(VarCharType.STRING_TYPE, new IntType()), + new MapType(VarCharType.STRING_TYPE, new VariantType()), + false, + true), + Arguments.of( + new ArrayType( + new YearMonthIntervalType( + YearMonthIntervalType.YearMonthResolution.MONTH)), + new ArrayType(new VariantType()), + false, + false), + // a whole constructed value does not cast into a single VARIANT yet + Arguments.of(new ArrayType(new IntType()), new VariantType(), false, false), + Arguments.of( + new MapType(VarCharType.STRING_TYPE, new IntType()), + new VariantType(), + false, + false), + Arguments.of( + new RowType(List.of(new RowField("a", new IntType()))), + new VariantType(), + false, + false)); } @ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Implicit: {2}, Explicit: {3}]") diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java index bd5ba1067a8..b1cfeaa1142 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java @@ -304,18 +304,16 @@ public class SqlCastFunction extends SqlFunction { SqlTypeName fromTypeName = fromType.getSqlTypeName(); SqlTypeName toTypeName = toType.getSqlTypeName(); - // Cast to Variant is not support at the moment. - // TODO: Support cast to variant (FLINK-37925，FLINK-37926) - if (toTypeName == SqlTypeName.VARIANT) { - return false; - } // Cast to BITMAP is not supported at the moment. if (toType instanceof BitmapRelDataType) { return false; } - // UUID casts are governed entirely by our own checker in both directions, because Calcite - // only allows a UUID to be cast from another UUID. - if (toTypeName == SqlTypeName.UUID || fromTypeName == SqlTypeName.UUID) { + // UUID and casts to VARIANT are governed entirely by our own checker, because Calcite + // only allows a UUID to be cast from another UUID and does not know which types VARIANT + // can store. + if (toTypeName == SqlTypeName.UUID + || fromTypeName == SqlTypeName.UUID + || toTypeName == SqlTypeName.VARIANT) { return LogicalTypeCasts.supportsExplicitCast( FlinkTypeFactory.toLogicalType(fromType), FlinkTypeFactory.toLogicalType(toType)); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java index 532c9bca994..e3e07fd0941 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java @@ -101,6 +101,7 @@ public class CastRuleProvider { .addRule(VariantToArrayCastRule.INSTANCE) .addRule(VariantToRowCastRule.INSTANCE) .addRule(VariantToMapCastRule.INSTANCE) + .addRule(PrimitiveToVariantCastRule.INSTANCE) // Bitmap rules .addRule(BitmapToStringCastRule.INSTANCE) .addRule(BitmapToBinaryCastRule.INSTANCE) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java new file mode 100644 index 00000000000..12c32e7324b --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java @@ -0,0 +1,179 @@ +/* + * 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.casting; + +import org.apache.flink.table.runtime.functions.VariantCastUtils; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.types.variant.Variant; + +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall; + +/** + * Cast rule from a primitive type to {@link LogicalTypeRoot#VARIANT}. + * + * <p>The value keeps the kind of its SQL type. For example, a {@code BIGINT} is stored as a {@code + * BIGINT}, even when it would fit a smaller kind. A {@code NaN} or infinite {@code FLOAT} or {@code + * DOUBLE} is stored as is, although {@code PARSE_JSON} rejects it. Some types can fail for some + * values, see {@link #canFail}. + */ +class PrimitiveToVariantCastRule extends AbstractExpressionCodeGeneratorCastRule<Object, Variant> { + + static final PrimitiveToVariantCastRule INSTANCE = new PrimitiveToVariantCastRule(); + + /** A character takes up to 4 bytes in UTF-8, which a declared length counts as one. */ + private static final int MAX_UTF8_BYTES_PER_CHAR = 4; + + private PrimitiveToVariantCastRule() { + super( + CastRulePredicate.builder() + .predicate( + (input, target) -> + target.is(LogicalTypeRoot.VARIANT) + && isSupportedSource(input)) + .build()); + } + + private static boolean isSupportedSource(LogicalType inputType) { + switch (inputType.getTypeRoot()) { + case BOOLEAN: + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + case CHAR: + case VARCHAR: + case BINARY: + case VARBINARY: + case DATE: + case TIME_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + case UUID: + return true; + default: + return false; + } + } + + /** + * Returns whether a value of the input type can fail the cast, so that {@code TRY_CAST} returns + * {@code NULL} for it instead of failing. Two kinds of types can fail: + * + * <ul> + * <li>A {@code TIMESTAMP(p)} or {@code TIMESTAMP_LTZ(p)} with {@code p > 6} is stored with + * nanoseconds, which only cover 1677-09-21 to 2262-04-11. + * <li>A string or binary value over {@link VariantCastUtils#MAX_PAYLOAD_BYTES} does not fit + * into a {@code VARIANT}. Only a type whose declared length allows such a value can fail. + * </ul> + * + * <p>Every other type never fails. + * + * <p>The length check trusts the declared length. Flink does not enforce that length on values + * from a source, so a longer value can still reach the cast. {@code TRY_CAST} then fails for it + * instead of returning {@code NULL}. + */ + @Override + public boolean canFail(LogicalType inputLogicalType, LogicalType targetLogicalType) { + switch (inputLogicalType.getTypeRoot()) { + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return LogicalTypeChecks.getPrecision(inputLogicalType) + > VariantCastUtils.TIMESTAMP_PRECISION; + case CHAR: + case VARCHAR: + return (long) MAX_UTF8_BYTES_PER_CHAR + * LogicalTypeChecks.getLength(inputLogicalType) + > VariantCastUtils.MAX_PAYLOAD_BYTES; + case BINARY: + case VARBINARY: + return LogicalTypeChecks.getLength(inputLogicalType) + > VariantCastUtils.MAX_PAYLOAD_BYTES; + default: + return false; + } + } + + /* Example generated code for INT: + + result$0 = org.apache.flink.table.runtime.functions.VariantCastUtils.fromIntegral(int$0); + + */ + @Override + public String generateExpression( + CodeGeneratorCastRule.Context context, + String inputTerm, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + if (isTimestamp(inputLogicalType)) { + return staticCall( + VariantCastUtils.class, + helperName(inputLogicalType), + inputTerm, + LogicalTypeChecks.getPrecision(inputLogicalType)); + } + return staticCall(VariantCastUtils.class, helperName(inputLogicalType), inputTerm); + } + + private static boolean isTimestamp(LogicalType inputLogicalType) { + return inputLogicalType.is(LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) + || inputLogicalType.is(LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE); + } + + private static String helperName(LogicalType inputLogicalType) { + switch (inputLogicalType.getTypeRoot()) { + case BOOLEAN: + return "fromBoolean"; + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + return "fromIntegral"; + case FLOAT: + return "fromFloat"; + case DOUBLE: + return "fromDouble"; + case DECIMAL: + return "fromDecimal"; + case CHAR: + case VARCHAR: + return "fromString"; + case BINARY: + case VARBINARY: + return "fromBytes"; + case DATE: + return "fromDate"; + case TIME_WITHOUT_TIME_ZONE: + return "fromTime"; + case TIMESTAMP_WITHOUT_TIME_ZONE: + return "fromTimestamp"; + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return "fromTimestampLtz"; + case UUID: + return "fromUuid"; + default: + throw new IllegalArgumentException( + "Unsupported source type for casting to VARIANT: " + inputLogicalType); + } + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java index e67bc261696..d7f4dc4a2a0 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java @@ -31,6 +31,8 @@ import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; import org.apache.flink.types.Row; import org.apache.flink.types.bitmap.Bitmap; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; import java.math.BigDecimal; import java.nio.ByteBuffer; @@ -160,6 +162,7 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { specs.addAll(variantArrayCasts()); specs.addAll(variantRowCasts()); specs.addAll(variantMapCasts()); + specs.addAll(primitiveToVariantCasts()); return specs; } @@ -432,6 +435,280 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { TINYINT())); } + private static List<TestSetSpec> primitiveToVariantCasts() { + final VariantBuilder builder = Variant.newBuilder(); + final LocalDateTime nanos = LocalDateTime.parse("2026-09-25T10:15:30.123456789"); + final Instant instant = Instant.parse("2026-09-25T10:15:30.123Z"); + return List.of( + // A value keeps the kind of its SQL type, so an integer keeps its width, and a + // string is wrapped rather than parsed. + CastTestSpecBuilder.testCastTo(VARIANT()) + .fromCase(BOOLEAN(), true, builder.of(true)) + .fromCase(INT(), 42, builder.of(42)) + .fromCase(BIGINT(), 1L, builder.of(1L)) + .fromCase(BIGINT(), 10000000000L, builder.of(10000000000L)) + .fromCase(DOUBLE(), 1.5d, builder.of(1.5d)) + .fromCase( + DECIMAL(4, 2), + new BigDecimal("12.50"), + builder.of(new BigDecimal("12.50"))) + .fromCase(STRING(), "{\"a\":1}", builder.of("{\"a\":1}")) + .fromCase( + DATE(), + LocalDate.parse("2026-09-25"), + builder.of(LocalDate.parse("2026-09-25"))) + .fromCase(TIMESTAMP(9), nanos, builder.of(nanos)) + .fromCase(TIMESTAMP_LTZ(3), instant, builder.of(instant)) + .fromCase(UUID(), DEFAULT_UUID, builder.of(DEFAULT_UUID)) + .fromCase(INT(), null, null) + // a type without a VARIANT kind is rejected at validation + .failValidation(INTERVAL(MONTH()), Period.ofMonths(2)) + .build(), + TestSetSpec.forExpression("Cast a primitive to VARIANT and back") + .onFieldsWithData( + 42L, + new BigDecimal("12.50"), + "hello", + LocalTime.of(10, 15, 30), + nanos, + instant, + DEFAULT_UUID, + new byte[] {1, 2, 3}) + .andDataTypes( + BIGINT(), + DECIMAL(4, 2), + STRING(), + TIME(), + TIMESTAMP(9), + TIMESTAMP_LTZ(3), + UUID(), + BYTES()) + .testResult( + $("f0").cast(VARIANT()).cast(BIGINT()), + "CAST(CAST(f0 AS VARIANT) AS BIGINT)", + 42L, + BIGINT()) + .testResult( + $("f1").cast(VARIANT()).cast(DECIMAL(4, 2)), + "CAST(CAST(f1 AS VARIANT) AS DECIMAL(4, 2))", + new BigDecimal("12.50"), + DECIMAL(4, 2)) + .testResult( + $("f2").cast(VARIANT()).cast(STRING()), + "CAST(CAST(f2 AS VARIANT) AS STRING)", + "hello", + STRING()) + .testResult( + $("f3").cast(VARIANT()).cast(TIME()), + "CAST(CAST(f3 AS VARIANT) AS TIME)", + LocalTime.of(10, 15, 30), + TIME()) + .testResult( + $("f4").cast(VARIANT()).cast(TIMESTAMP(9)), + "CAST(CAST(f4 AS VARIANT) AS TIMESTAMP(9))", + nanos, + TIMESTAMP(9)) + .testResult( + $("f5").cast(VARIANT()).cast(TIMESTAMP_LTZ(3)), + "CAST(CAST(f5 AS VARIANT) AS TIMESTAMP_LTZ(3))", + instant, + TIMESTAMP_LTZ(3)) + .testResult( + $("f6").cast(VARIANT()).cast(UUID()), + "CAST(CAST(f6 AS VARIANT) AS UUID)", + DEFAULT_UUID, + UUID()) + .testResult( + $("f7").cast(VARIANT()).cast(BYTES()), + "CAST(CAST(f7 AS VARIANT) AS BYTES)", + new byte[] {1, 2, 3}, + BYTES()), + TestSetSpec.forExpression("Cast TIME to VARIANT and back") + .onFieldsWithData( + LocalTime.of(12, 34, 56, 123_000_000), + LocalTime.MIDNIGHT, + LocalTime.of(23, 59, 59, 999_000_000)) + .andDataTypes(TIME(3), TIME(0), TIME(3)) + .testResult( + $("f0").cast(VARIANT()).cast(TIME(3)), + "CAST(CAST(f0 AS VARIANT) AS TIME(3))", + LocalTime.of(12, 34, 56, 123_000_000), + TIME(3)) + .testResult( + $("f1").cast(VARIANT()).cast(TIME(0)), + "CAST(CAST(f1 AS VARIANT) AS TIME(0))", + LocalTime.MIDNIGHT, + TIME(0)) + .testResult( + $("f2").cast(VARIANT()).cast(TIME(3)), + "CAST(CAST(f2 AS VARIANT) AS TIME(3))", + LocalTime.of(23, 59, 59, 999_000_000), + TIME(3)), + // a VARIANT holds at most 16 MiB, so TRY_CAST returns NULL for a longer string + TestSetSpec.forExpression("Cast a string over the size limit to VARIANT") + .onFieldsWithData("x") + .andDataTypes(STRING()) + .testSqlResult( + "TRY_CAST(REPEAT(f0, 17000000) AS VARIANT)", null, VARIANT()), + // nanoseconds only cover 1677-09-21 to 2262-04-11, while microseconds hold any year + TestSetSpec.forExpression("Cast a late TIMESTAMP to VARIANT") + .onFieldsWithData( + LocalDateTime.parse("3000-01-01T00:00"), + LocalDateTime.parse("3000-01-01T00:00")) + .andDataTypes(TIMESTAMP(9), TIMESTAMP(6)) + .testSqlRuntimeError( + "CAST(f0 AS VARIANT)", + TableRuntimeException.class, + "1677-09-21 to 2262-04-11") + .testTableApiRuntimeError( + $("f0").cast(VARIANT()), + TableRuntimeException.class, + "1677-09-21 to 2262-04-11") + .testResult( + $("f0").tryCast(VARIANT()), + "TRY_CAST(f0 AS VARIANT)", + null, + VARIANT()) + .testResult( + $("f1").cast(VARIANT()).cast(TIMESTAMP(6)), + "CAST(CAST(f1 AS VARIANT) AS TIMESTAMP(6))", + LocalDateTime.parse("3000-01-01T00:00"), + TIMESTAMP(6)), + TestSetSpec.forExpression("Cast a literal to VARIANT") + .onFieldsWithData(0) + .testResult( + lit(42).cast(VARIANT()), + "CAST(42 AS VARIANT)", + builder.of(42), + VARIANT().notNull()) + .testResult( + lit("hello").cast(VARIANT()), + "CAST('hello' AS VARIANT)", + builder.of("hello"), + VARIANT().notNull()) + // a NULL literal casts to a SQL NULL, not to a variant null + .testSqlResult("CAST(NULL AS VARIANT)", null, VARIANT()) + .testSqlResult("TRY_CAST(NULL AS VARIANT)", null, VARIANT()), + // a VARIANT is not limited to JSON, so NaN and infinity are kept and round-trip + TestSetSpec.forExpression("Cast a NaN literal to VARIANT and back") + .onFieldsWithData(0) + .testResult( + lit("NaN").cast(DOUBLE()).cast(VARIANT()).cast(DOUBLE()), + "CAST(CAST(CAST('NaN' AS DOUBLE) AS VARIANT) AS DOUBLE)", + Double.NaN, + DOUBLE()) + .testResult( + lit("NaN").cast(DOUBLE()).cast(VARIANT()).cast(STRING()), + "CAST(CAST(CAST('NaN' AS DOUBLE) AS VARIANT) AS STRING)", + "NaN", + STRING()), + TestSetSpec.forExpression("Cast a non-finite number to VARIANT and back") + // the non-finite values are produced at runtime from strings + .onFieldsWithData("NaN", "Infinity", "-Infinity") + .andDataTypes(STRING(), STRING(), STRING()) + .testResult( + $("f0").cast(DOUBLE()).cast(VARIANT()).cast(DOUBLE()), + "CAST(CAST(CAST(f0 AS DOUBLE) AS VARIANT) AS DOUBLE)", + Double.NaN, + DOUBLE()) + .testResult( + $("f1").cast(FLOAT()).cast(VARIANT()).cast(FLOAT()), + "CAST(CAST(CAST(f1 AS FLOAT) AS VARIANT) AS FLOAT)", + Float.POSITIVE_INFINITY, + FLOAT()) + .testResult( + $("f2").cast(DOUBLE()).cast(VARIANT()).cast(FLOAT()), + "CAST(CAST(CAST(f2 AS DOUBLE) AS VARIANT) AS FLOAT)", + Float.NEGATIVE_INFINITY, + FLOAT()) + // an integer cannot hold NaN, so that cast still fails + .testSqlRuntimeError( + "CAST(CAST(CAST(f0 AS DOUBLE) AS VARIANT) AS INT)", + TableRuntimeException.class, + "overflowed"), + TestSetSpec.forExpression("Cast narrow and fixed-width types to VARIANT and back") + .onFieldsWithData((byte) 42, (short) 1000) + .andDataTypes(TINYINT(), SMALLINT()) + .testResult( + $("f0").cast(VARIANT()).cast(TINYINT()), + "CAST(CAST(f0 AS VARIANT) AS TINYINT)", + (byte) 42, + TINYINT()) + .testResult( + $("f1").cast(VARIANT()).cast(SMALLINT()), + "CAST(CAST(f1 AS VARIANT) AS SMALLINT)", + (short) 1000, + SMALLINT()) + // a CHAR(n) is stored with its padding, so the spaces survive + .testResult( + lit("ab").cast(CHAR(4)).cast(VARIANT()).cast(STRING()), + "CAST(CAST(CAST('ab' AS CHAR(4)) AS VARIANT) AS STRING)", + "ab ", + STRING()) + .testResult( + lit("ab").cast(CHAR(4)).cast(VARIANT()).cast(CHAR(4)), + "CAST(CAST(CAST('ab' AS CHAR(4)) AS VARIANT) AS CHAR(4))", + "ab ", + CHAR(4)) + // a BINARY(n) is stored with its zero padding + .testResult( + lit(new byte[] {1, 2}) + .cast(BINARY(4)) + .cast(VARIANT()) + .cast(BYTES()), + "CAST(CAST(CAST(X'0102' AS BINARY(4)) AS VARIANT) AS BYTES)", + new byte[] {1, 2, 0, 0}, + BYTES()), + // a constructed type casts element by element when each child casts to VARIANT + TestSetSpec.forExpression("Cast a constructed type to VARIANT elements and back") + .onFieldsWithData( + new Integer[] {1, null}, + Row.of(7), + map(entry("a", 1)), + map(entry(1, "a"))) + .andDataTypes( + ARRAY(INT()), + ROW(FIELD("a", INT())), + MAP(STRING(), INT()), + MAP(INT(), STRING())) + .testResult( + $("f0").cast(ARRAY(VARIANT())).cast(ARRAY(INT())), + "CAST(CAST(f0 AS ARRAY<VARIANT>) AS ARRAY<INT>)", + new Integer[] {1, null}, + ARRAY(INT())) + .testResult( + $("f1").cast(ROW(FIELD("a", VARIANT()))) + .cast(ROW(FIELD("a", INT()))), + "CAST(CAST(f1 AS ROW<a VARIANT>) AS ROW<a INT>)", + Row.of(7), + ROW(FIELD("a", INT()))) + .testResult( + $("f2").cast(MAP(STRING(), VARIANT())).cast(MAP(STRING(), INT())), + "CAST(CAST(f2 AS MAP<STRING, VARIANT>) AS MAP<STRING, INT>)", + map(entry("a", 1)), + MAP(STRING(), INT())) + // a VARIANT key matches by its bytes, so a key cast from INT does not + // match the same number cast from BIGINT + .testResult( + $("f3").cast(MAP(VARIANT(), STRING())).at(lit(1).cast(VARIANT())), + "CAST(f3 AS MAP<VARIANT, STRING>)[CAST(1 AS VARIANT)]", + "a", + STRING()) + .testResult( + $("f3").cast(MAP(VARIANT(), STRING())).at(lit(1L).cast(VARIANT())), + "CAST(f3 AS MAP<VARIANT, STRING>)[CAST(CAST(1 AS BIGINT) AS VARIANT)]", + null, + STRING()), + // a whole constructed value does not cast into a single VARIANT yet + TestSetSpec.forExpression("Cast a constructed type to VARIANT") + .onFieldsWithData(0) + .testSqlValidationError( + "CAST(ARRAY[1, 2] AS VARIANT)", + "Cast function cannot convert value") + .testTableApiValidationError( + lit(new int[] {1, 2}).cast(VARIANT()), "Unsupported cast")); + } + private static List<TestSetSpec> variantArrayCasts() { return List.of( TestSetSpec.forExpression("Cast a VARIANT produced by parseJson() to an ARRAY") diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java index df910dfe1c4..6849d79e6ad 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java @@ -32,9 +32,12 @@ import static org.apache.flink.table.api.DataTypes.ARRAY; import static org.apache.flink.table.api.DataTypes.BIGINT; import static org.apache.flink.table.api.DataTypes.BOOLEAN; import static org.apache.flink.table.api.DataTypes.BYTES; +import static org.apache.flink.table.api.DataTypes.CHAR; import static org.apache.flink.table.api.DataTypes.DATE; import static org.apache.flink.table.api.DataTypes.DECIMAL; +import static org.apache.flink.table.api.DataTypes.DOUBLE; import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.api.DataTypes.FLOAT; import static org.apache.flink.table.api.DataTypes.INT; import static org.apache.flink.table.api.DataTypes.INTERVAL; import static org.apache.flink.table.api.DataTypes.MAP; @@ -48,6 +51,8 @@ import static org.apache.flink.table.api.DataTypes.TIMESTAMP; import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ; import static org.apache.flink.table.api.DataTypes.TINYINT; import static org.apache.flink.table.api.DataTypes.UUID; +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.DataTypes.VARIANT; import static org.apache.flink.table.types.logical.VarCharType.STRING_TYPE; import static org.assertj.core.api.Assertions.assertThat; @@ -147,6 +152,74 @@ class CastRuleProviderTest { .isSameAs(VariantToStringCastRule.INSTANCE); } + @Test + void testResolvePrimitiveToVariant() { + assertThat(CastRuleProvider.resolve(INT, VARIANT)) + .isSameAs(PrimitiveToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(STRING_TYPE, VARIANT)) + .isSameAs(PrimitiveToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.exists(DECIMAL(10, 2).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.exists(TIMESTAMP(9).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.exists(TIMESTAMP_LTZ().getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.exists(TIME().getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.exists(BYTES().getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.exists(UUID().getLogicalType(), VARIANT)).isTrue(); + + // only a nanosecond timestamp can fail, since that kind covers a limited range of years + assertThat(CastRuleProvider.canFail(INT, VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(DOUBLE().getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(FLOAT().getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(TIMESTAMP(6).getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(TIMESTAMP(9).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(TIMESTAMP_LTZ(3).getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(TIMESTAMP_LTZ(7).getLogicalType(), VARIANT)).isTrue(); + + // a string or binary type fails only when its declared length allows a value over 16 MiB, + // and a character takes up to 4 bytes in UTF-8 + assertThat(CastRuleProvider.canFail(STRING_TYPE, VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(BYTES().getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(VARCHAR(100).getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(CHAR(100).getLogicalType(), VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(VARCHAR(4_194_302).getLogicalType(), VARIANT)) + .isFalse(); + assertThat(CastRuleProvider.canFail(VARCHAR(4_194_303).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(VARBINARY(16_777_211).getLogicalType(), VARIANT)) + .isFalse(); + assertThat(CastRuleProvider.canFail(VARBINARY(16_777_212).getLogicalType(), VARIANT)) + .isTrue(); + + // INTERVAL has no VARIANT kind, so it is not a castable source + assertThat(CastRuleProvider.exists(INTERVAL(DataTypes.DAY()).getLogicalType(), VARIANT)) + .isFalse(); + // VARIANT to VARIANT stays the identity + assertThat(CastRuleProvider.resolve(VARIANT, VARIANT)).isSameAs(IdentityCastRule.INSTANCE); + + // a constructed type casts element by element, through the rules for its children + assertThat( + CastRuleProvider.resolve( + ARRAY(INT()).getLogicalType(), ARRAY(VARIANT()).getLogicalType())) + .isSameAs(ArrayToArrayCastRule.INSTANCE); + assertThat( + CastRuleProvider.resolve( + ROW(FIELD("a", INT())).getLogicalType(), + ROW(FIELD("a", VARIANT())).getLogicalType())) + .isSameAs(RowToRowCastRule.INSTANCE); + assertThat( + CastRuleProvider.resolve( + MAP(STRING(), INT()).getLogicalType(), + MAP(STRING(), VARIANT()).getLogicalType())) + .isSameAs(MapToMapAndMultisetToMultisetCastRule.INSTANCE); + assertThat( + CastRuleProvider.canFail( + ARRAY(INT()).getLogicalType(), ARRAY(VARIANT()).getLogicalType())) + .isFalse(); + assertThat( + CastRuleProvider.canFail( + ARRAY(TIMESTAMP(9)).getLogicalType(), + ARRAY(VARIANT()).getLogicalType())) + .isTrue(); + } + @Test void testResolveVariantToArray() { assertThat(CastRuleProvider.resolve(VARIANT, ARRAY(INT()).getLogicalType())) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index cfe2b641bfd..33015cef9e1 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -42,6 +42,7 @@ import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.utils.DateTimeUtils; import org.apache.flink.types.bitmap.Bitmap; import org.apache.flink.types.variant.BinaryVariant; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; @@ -60,6 +61,8 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; import java.util.AbstractMap; import java.util.ArrayList; import java.util.Arrays; @@ -1946,6 +1949,165 @@ class CastRulesTest { VARIANT_BUILDER.of(NON_BMP + NON_BMP + NON_BMP), fromString(NON_BMP + NON_BMP)) .fromCase(VARIANT(), VARIANT_BUILDER.of("a"), fromString("a")), + CastTestSpecBuilder.testCastTo(VARIANT()) + .fromCase(BOOLEAN(), true, VARIANT_BUILDER.of(true)) + // an integer keeps the width of its SQL type, even when a smaller one fits + .fromCase(TINYINT(), (byte) 42, VARIANT_BUILDER.of((byte) 42)) + .fromCase(SMALLINT(), (short) 42, VARIANT_BUILDER.of((short) 42)) + .fromCase(INT(), 42, VARIANT_BUILDER.of(42)) + .fromCase(BIGINT(), 1L, VARIANT_BUILDER.of(1L)) + .fromCase(BIGINT(), 42L, VARIANT_BUILDER.of(42L)) + .fromCase( + BIGINT(), + DEFAULT_NEGATIVE_BIGINT, + VARIANT_BUILDER.of(DEFAULT_NEGATIVE_BIGINT)) + .fromCase(BIGINT(), Long.MAX_VALUE, VARIANT_BUILDER.of(Long.MAX_VALUE)) + .fromCase(FLOAT(), 1.5f, VARIANT_BUILDER.of(1.5f)) + .fromCase(DOUBLE(), 1.5d, VARIANT_BUILDER.of(1.5d)) + .fromCase(DOUBLE(), Double.MAX_VALUE, VARIANT_BUILDER.of(Double.MAX_VALUE)) + // a VARIANT is not limited to JSON, so NaN and infinity are kept + .fromCase(FLOAT(), Float.NaN, VARIANT_BUILDER.of(Float.NaN)) + .fromCase( + FLOAT(), + Float.POSITIVE_INFINITY, + VARIANT_BUILDER.of(Float.POSITIVE_INFINITY)) + .fromCase(DOUBLE(), Double.NaN, VARIANT_BUILDER.of(Double.NaN)) + .fromCase( + DOUBLE(), + Double.NEGATIVE_INFINITY, + VARIANT_BUILDER.of(Double.NEGATIVE_INFINITY)) + // a decimal keeps its scale + .fromCase( + DECIMAL(4, 2), + fromBigDecimal(new BigDecimal("12.50"), 4, 2), + VARIANT_BUILDER.of(new BigDecimal("12.50"))) + .fromCase( + DECIMAL(38, 0), + fromBigDecimal(new BigDecimal("9".repeat(38)), 38, 0), + VARIANT_BUILDER.of(new BigDecimal("9".repeat(38)))) + // a string is wrapped, never parsed as JSON + .fromCase( + STRING(), fromString("{\"a\":1}"), VARIANT_BUILDER.of("{\"a\":1}")) + .fromCase(CHAR(4), fromString("ab "), VARIANT_BUILDER.of("ab ")) + .fromCase( + BYTES(), + new byte[] {1, 2, 3}, + VARIANT_BUILDER.of(new byte[] {1, 2, 3})) + .fromCase(DATE(), DATE, VARIANT_BUILDER.of(LocalDate.parse("2021-09-24"))) + // VARIANT stores TIME in microseconds, so every Flink TIME fits exactly, + // since the runtime keeps milliseconds + .fromCase(TIME(), TIME, VARIANT_BUILDER.of(LocalTime.parse("12:34:56.123"))) + .fromCase( + TIME(0), + DateTimeUtils.toInternal(LocalTime.of(12, 34, 56)), + VARIANT_BUILDER.of(LocalTime.of(12, 34, 56))) + .fromCase( + TIME(3), + DateTimeUtils.toInternal(LocalTime.of(12, 34, 56, 123_000_000)), + VARIANT_BUILDER.of(LocalTime.of(12, 34, 56, 123_000_000))) + .fromCase(TIME(0), 0, VARIANT_BUILDER.of(LocalTime.MIDNIGHT)) + .fromCase( + TIME(3), + DateTimeUtils.toInternal(LocalTime.of(23, 59, 59, 999_000_000)), + VARIANT_BUILDER.of(LocalTime.of(23, 59, 59, 999_000_000))) + // microsecond precision and below is stored as a TIMESTAMP + .fromCase( + TIMESTAMP(3), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("2021-09-24T12:34:56.123")), + VARIANT_BUILDER.of(LocalDateTime.parse("2021-09-24T12:34:56.123"))) + .fromCase( + TIMESTAMP(6), + TIMESTAMP, + VARIANT_BUILDER.of( + LocalDateTime.parse("2021-09-24T12:34:56.123456"))) + // micros before the epoch are negative, 1969-12-31T23:59:59.999999 is -1 + .fromCase( + TIMESTAMP(6), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("1969-12-31T23:59:59.999999")), + VARIANT_BUILDER.of( + LocalDateTime.parse("1969-12-31T23:59:59.999999"))) + .fromCase( + TIMESTAMP_LTZ(6), + TimestampData.fromInstant( + Instant.parse("1969-12-31T23:59:59.999999Z")), + VARIANT_BUILDER.of(Instant.parse("1969-12-31T23:59:59.999999Z"))) + // a microsecond precision holds any year, so a late date still casts + .fromCase( + TIMESTAMP(6), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("3000-01-01T00:00")), + VARIANT_BUILDER.of(LocalDateTime.parse("3000-01-01T00:00"))) + // a precision above 6 is stored with nanoseconds, even for a value that has + // no digits below a microsecond + .fromCase( + TIMESTAMP(9), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("2021-09-24T12:34:56.123456789")), + VARIANT_BUILDER.of( + LocalDateTime.parse("2021-09-24T12:34:56.123456789"))) + .fromCase( + TIMESTAMP(9), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("2021-09-24T12:34:56.123456")), + timestampNanosVariant( + LocalDateTime.parse("2021-09-24T12:34:56.123456"))) + .fromCase( + TIMESTAMP_LTZ(3), + TimestampData.fromInstant( + Instant.parse("2022-01-04T12:34:56.123Z")), + VARIANT_BUILDER.of(Instant.parse("2022-01-04T12:34:56.123Z"))) + .fromCase( + TIMESTAMP_LTZ(9), + TIMESTAMP_LTZ, + VARIANT_BUILDER.of(TIMESTAMP_LTZ.toInstant())) + .fromCase( + TIMESTAMP_LTZ(9), + TimestampData.fromInstant(Instant.parse("2022-01-04T12:34:56Z")), + timestampLtzNanosVariant(Instant.parse("2022-01-04T12:34:56Z"))) + // nanoseconds only cover 1677-09-21 to 2262-04-11 + .fail( + TIMESTAMP(9), + TimestampData.fromLocalDateTime( + LocalDateTime.parse("3000-01-01T00:00")), + TableRuntimeException.class, + "1677-09-21 to 2262-04-11") + .fail( + TIMESTAMP_LTZ(7), + TimestampData.fromInstant(Instant.parse("1500-01-01T00:00:00Z")), + TableRuntimeException.class, + "1677-09-21 to 2262-04-11") + .fromCase(UUID(), UUID_BYTES, VARIANT_BUILDER.of(UUID_VALUE)) + // a SQL NULL stays a SQL NULL rather than becoming a variant null + .fromCase(INT(), null, null) + .fromCase(STRING(), null, null), + // a constructed type casts element by element, and a NULL element stays SQL NULL + CastTestSpecBuilder.testCastTo(ARRAY(VARIANT())) + .fromCase( + ARRAY(INT()), + new GenericArrayData(new Integer[] {1, null}), + new GenericArrayData(new Object[] {VARIANT_BUILDER.of(1), null})), + CastTestSpecBuilder.testCastTo(ROW(FIELD("a", VARIANT()), FIELD("b", VARIANT()))) + .fromCase( + ROW(FIELD("a", INT()), FIELD("b", STRING())), + GenericRowData.of(7, null), + GenericRowData.of(VARIANT_BUILDER.of(7), null)), + CastTestSpecBuilder.testCastTo(MAP(STRING(), VARIANT())) + .fromCase( + MAP(STRING(), INT()), + mapData(entry(fromString("a"), 1)), + mapData(entry(fromString("a"), VARIANT_BUILDER.of(1)))), + CastTestSpecBuilder.testCastTo(MAP(VARIANT(), STRING())) + .fromCase( + MAP(INT(), STRING()), + mapData(entry(1, fromString("a"))), + mapData(entry(VARIANT_BUILDER.of(1), fromString("a")))), + CastTestSpecBuilder.testCastTo(MULTISET(VARIANT())) + .fromCase( + MULTISET(INT()), + mapData(entry(1, 2)), + mapData(entry(VARIANT_BUILDER.of(1), 2))), CastTestSpecBuilder.testCastTo(BOOLEAN()) .fromCase(VARIANT(), VARIANT_BUILDER.of(true), true) .fromCase(VARIANT(), VARIANT_BUILDER.of(false), false) @@ -2766,6 +2928,20 @@ class CastRulesTest { LocalDateTime.of(years, months, days, hours, minutes, seconds, nanos)); } + /** A nanosecond timestamp, which {@link VariantBuilder} only builds for sub-micro digits. */ + private static Variant timestampNanosVariant(LocalDateTime value) { + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + builder.appendTimestampNanos( + ChronoUnit.NANOS.between(Instant.EPOCH, value.toInstant(ZoneOffset.UTC))); + return builder.build(); + } + + private static Variant timestampLtzNanosVariant(Instant value) { + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + builder.appendTimestampLtzNanos(ChronoUnit.NANOS.between(Instant.EPOCH, value)); + return builder.build(); + } + private static TimestampData timestampDataFromInstant( int years, int months, int days, int hours, int minutes, int seconds, int nanos) { return TimestampData.fromInstant( diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index 9f2deeb71d4..b834a157c76 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.functions; import org.apache.flink.annotation.Internal; import org.apache.flink.table.api.TableRuntimeException; import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.data.binary.BinaryStringData; import org.apache.flink.table.data.binary.BinaryStringDataUtil; @@ -28,18 +29,22 @@ import org.apache.flink.table.data.binary.StringUtf8Utils; import org.apache.flink.table.types.logical.utils.UuidUtils; import org.apache.flink.table.utils.DateTimeUtils; import org.apache.flink.table.utils.EncodingUtils; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; import org.apache.flink.types.variant.BinaryVariantUtil; import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; import org.apache.flink.types.variant.VariantTypeException; import java.math.BigDecimal; import java.math.RoundingMode; import java.nio.charset.StandardCharsets; +import java.time.LocalDate; +import java.time.LocalTime; import java.util.Arrays; import java.util.TimeZone; /** - * Runtime helpers for casting a {@code VARIANT} value to a SQL type. + * Runtime helpers for casting between a {@code VARIANT} value and a SQL type. * * <p>A cast never reinterprets the stored value: one kind is not read as another, and a numeric * value is never wrapped or rounded to make it fit. Any numeric kind therefore reaches an integer @@ -51,6 +56,9 @@ import java.util.TimeZone; * regular cast into the same type. A value longer than the target is trimmed, fractional seconds * beyond the target precision are truncated, and the fixed width targets {@code CHAR(n)} and {@code * BINARY(n)} pad a shorter value. + * + * <p>A cast into {@code VARIANT} keeps the kind of the SQL type, so an integer keeps its width and + * a timestamp keeps the precision it was declared with. */ @Internal public final class VariantCastUtils { @@ -62,7 +70,15 @@ public final class VariantCastUtils { */ private static final double LONG_MAGNITUDE_LIMIT = -(double) Long.MIN_VALUE; - private static final int TIMESTAMP_PRECISION = 6; + /** The highest timestamp precision a {@code VARIANT} stores with microseconds. */ + public static final int TIMESTAMP_PRECISION = 6; + + /** + * The largest string or binary value a {@code VARIANT} holds, in bytes. A {@code VARIANT} is + * limited to 16 MiB, and a long string or binary value spends 5 bytes of that on its header. + */ + public static final int MAX_PAYLOAD_BYTES = + BinaryVariantUtil.SIZE_LIMIT - 1 - BinaryVariantUtil.U32_SIZE; private static final int TIMESTAMP_NANOS_PRECISION = 9; @@ -71,6 +87,8 @@ public final class VariantCastUtils { private static final String NULL_NODE = "NULL"; + private static final VariantBuilder BUILDER = Variant.newBuilder(); + private VariantCastUtils() {} /** @@ -541,6 +559,131 @@ public final class VariantCastUtils { } } + public static Variant fromBoolean(boolean value) { + return BUILDER.of(value); + } + + /** + * Stores an integer in the integer kind of its SQL type, so a {@code BIGINT} stays a {@code + * BIGINT} whatever its value. The generated code passes the primitive of the source type, which + * selects the overload. + */ + public static Variant fromIntegral(byte value) { + return BUILDER.of(value); + } + + public static Variant fromIntegral(short value) { + return BUILDER.of(value); + } + + public static Variant fromIntegral(int value) { + return BUILDER.of(value); + } + + public static Variant fromIntegral(long value) { + return BUILDER.of(value); + } + + public static Variant fromFloat(float value) { + return BUILDER.of(value); + } + + public static Variant fromDouble(double value) { + return BUILDER.of(value); + } + + public static Variant fromDecimal(DecimalData value) { + return BUILDER.of(value.toBigDecimal()); + } + + public static Variant fromString(StringData value) { + try { + return BUILDER.of(value.toString()); + } catch (VariantTypeException e) { + throw sizeLimitExceeded(e, "string", value.toBytes().length); + } + } + + public static Variant fromBytes(byte[] value) { + try { + return BUILDER.of(value); + } catch (VariantTypeException e) { + throw sizeLimitExceeded(e, "binary", value.length); + } + } + + private static RuntimeException sizeLimitExceeded( + VariantTypeException e, String kind, int bytes) { + if (e != BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION) { + return e; + } + return new TableRuntimeException( + String.format( + "Cannot cast a %s value of %d bytes to VARIANT. A VARIANT is limited to " + + "16 MiB, so a string or binary value can have at most %d bytes.", + kind, bytes, MAX_PAYLOAD_BYTES)); + } + + public static Variant fromDate(int epochDay) { + return BUILDER.of(LocalDate.ofEpochDay(epochDay)); + } + + public static Variant fromTime(int millisOfDay) { + return BUILDER.of(LocalTime.ofNanoOfDay(millisOfDay * 1_000_000L)); + } + + /** + * Stores a timestamp in the kind its declared precision needs: microseconds up to {@code + * TIMESTAMP(6)} and nanoseconds above, whatever the digits of the value. A nanosecond kind only + * covers 1677-09-21 to 2262-04-11, so a {@code TIMESTAMP(7)} to {@code TIMESTAMP(9)} value + * outside that range fails. + */ + public static Variant fromTimestamp(TimestampData value, int precision) { + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + if (precision <= TIMESTAMP_PRECISION) { + builder.appendTimestamp(timestampMicros(value)); + } else { + builder.appendTimestampNanos(timestampNanos(value, "TIMESTAMP(" + precision + ")")); + } + return builder.build(); + } + + /** Like {@link #fromTimestamp(TimestampData, int)}, for {@code TIMESTAMP_LTZ}. */ + public static Variant fromTimestampLtz(TimestampData value, int precision) { + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + if (precision <= TIMESTAMP_PRECISION) { + builder.appendTimestampLtz(timestampMicros(value)); + } else { + builder.appendTimestampLtzNanos( + timestampNanos(value, "TIMESTAMP_LTZ(" + precision + ")")); + } + return builder.build(); + } + + /** Microseconds since the epoch, which cover every year a {@link TimestampData} can hold. */ + private static long timestampMicros(TimestampData value) { + return value.getMillisecond() * 1_000L + value.getNanoOfMillisecond() / 1_000; + } + + private static long timestampNanos(TimestampData value, String sourceType) { + try { + return Math.addExact( + Math.multiplyExact(value.getMillisecond(), 1_000_000L), + value.getNanoOfMillisecond()); + } catch (ArithmeticException e) { + throw new TableRuntimeException( + String.format( + "Cannot cast the %s value %s to VARIANT. A VARIANT timestamp with " + + "nanosecond precision only covers 1677-09-21 to 2262-04-11. " + + "Cast the value to a precision of 6 or less first.", + sourceType, value)); + } + } + + public static Variant fromUuid(byte[] value) { + return BUILDER.of(UuidUtils.fromBytes(value)); + } + private static TableRuntimeException unsupportedKind(Variant variant, String targetType) { return new TableRuntimeException( String.format( diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java index f07452165d3..4492a94b234 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java @@ -18,6 +18,8 @@ package org.apache.flink.table.runtime.functions; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.data.StringData; import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; @@ -27,10 +29,14 @@ import org.junit.jupiter.api.Test; import java.time.Instant; import java.util.TimeZone; +import static org.apache.flink.table.runtime.functions.VariantCastUtils.MAX_PAYLOAD_BYTES; +import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromBytes; +import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromString; import static org.apache.flink.table.runtime.functions.VariantCastUtils.toPrintString; import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader; import static org.apache.flink.types.variant.BinaryVariantUtil.shortStrHeader; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class VariantCastUtilsTest { @@ -38,6 +44,27 @@ class VariantCastUtilsTest { private static final VariantBuilder BUILDER = Variant.newBuilder(); + @Test + void testCastToVariantHoldsUpToTheSizeLimit() { + assertThat(fromBytes(new byte[MAX_PAYLOAD_BYTES]).getBytes()).hasSize(MAX_PAYLOAD_BYTES); + assertThat(fromString(StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES))).getString()) + .hasSize(MAX_PAYLOAD_BYTES); + } + + @Test + void testCastToVariantRejectsValuesOverTheSizeLimit() { + assertThatThrownBy(() -> fromBytes(new byte[MAX_PAYLOAD_BYTES + 1])) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a binary value of 16777212 bytes to VARIANT. A VARIANT is " + + "limited to 16 MiB, so a string or binary value can have at " + + "most 16777211 bytes."); + assertThatThrownBy( + () -> fromString(StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES + 1)))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); + } + @Test void testPrintRendersLikeCast() { final Variant variant =
