This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 140140a88fbf6e09ae76d37dd8da135c61e6cd1b Author: Ramin Gharib <[email protected]> AuthorDate: Tue Oct 6 07:53:41 2026 +0200 [FLINK-40826][table] Support casting ARRAY, MAP, ROW and STRUCTURED to VARIANT CAST and TRY_CAST now turn a whole constructed value into one VARIANT. An ARRAY becomes a variant array, and a MAP, ROW or STRUCTURED value a variant object keyed by its field names or its character string keys. A NULL element, field or map value becomes a variant null, and a nested VARIANT is copied in as is. Leaves are stored like the cast to VARIANT stores them on their own. A NULL map key fails the cast, and so does a value over the 16 MiB of a VARIANT or a VARIANT nested too deep [...] LogicalTypeCasts allows these casts when every leaf casts to VARIANT and every MAP key is a character string. ToVariantCastRule accepts them, and its canFail treats every ARRAY, MAP and nested VARIANT as fallible and bounds the size of a ROW by the declared sizes of its fields. --- docs/content/docs/sql/reference/data-types.md | 46 +++++- .../types/logical/utils/LogicalTypeCasts.java | 17 +++ .../flink/table/types/LogicalTypeCastsTest.java | 64 +++++++- .../functions/casting/ToVariantCastRule.java | 89 ++++++++--- .../planner/functions/CastFunctionITCase.java | 170 ++++++++++++++++++++- .../functions/casting/CastRuleProviderTest.java | 47 ++++++ .../planner/functions/casting/CastRulesTest.java | 87 ++++++++++- 7 files changed, 484 insertions(+), 36 deletions(-) diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index c2c94bb2c5c..c80cb694fc2 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1714,7 +1714,7 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a `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. + whole constructed value can also become a single `VARIANT`, see below. - 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')`. @@ -1727,7 +1727,40 @@ 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 +``` + +A whole `ARRAY`, `MAP`, `ROW`, or `STRUCTURED` value can also be cast into a single `VARIANT`. An +`ARRAY` becomes a variant array, and a `MAP`, `ROW`, or `STRUCTURED` value a variant object. Each +leaf is stored by the rules above, so the cast is supported only when every leaf type casts to +`VARIANT`. + +- A `ROW` or `STRUCTURED` value is keyed by its field names. +- A `MAP` needs a character string key, which becomes the object key. A `NULL` key fails the cast. + If a key appears twice, the last value is kept. +- A variant object sorts its keys, so neither the field order of a `ROW` nor the entry order of a + `MAP` is kept. +- A `NULL` element, field, or map value becomes a variant null, so an array keeps its length and an + object keeps its keys. A cast element by element differs here: `ARRAY<INT>` to `ARRAY<VARIANT>` + keeps a SQL `NULL`, while `ARRAY<INT>` to `VARIANT` turns it into a variant null. +- A nested `VARIANT` is embedded as is. +- The whole value must fit into the 16 MiB of a `VARIANT`. Any `ARRAY` or `MAP` can exceed it, and so + can a `ROW` with a nested `VARIANT` or with fields whose declared sizes add up to more. The cast + then fails, and `TRY_CAST` returns `NULL` for the whole value. +- Casting the `VARIANT` back to the original type returns the original value, since a cast to `ROW` + matches fields by name. There are two exceptions: + - A SQL `NULL` in a `VARIANT` field, in an `ARRAY<VARIANT>`, or as a value of a + `MAP<STRING, VARIANT>` comes back as a variant null. + - A `MAP` that holds a key twice comes back with only the last value for that key. + +The following examples use `r` for a column of type `ROW<name STRING, id INT>` that holds +`('ada', 7)`: + +```sql +CAST(ARRAY[1, NULL] AS VARIANT) -- [1, null], the NULL becomes a variant null +CAST(MAP['b', 2, 'a', 1] AS VARIANT) -- {"a": 1, "b": 2} +CAST(r AS VARIANT) -- {"id": 7, "name": "ada"} +CAST(MAP[1, 'a'] AS VARIANT) -- fails at validation, a MAP key must be a character string +CAST(ARRAY[INTERVAL '1' DAY] AS VARIANT) -- fails at validation ``` **Declaration** @@ -1942,11 +1975,11 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | `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 | +| `ARRAY` | Y | 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 | -| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | -| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | -| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | +| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | !⁸ | N | +| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | !⁸ | N | +| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | !⁸ | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | | `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | @@ -1960,6 +1993,7 @@ Notes: 5. Supported iff `INTERVAL` is a `MONTH TO YEAR` range. 6. Supported iff `INTERVAL` is a `DAY TO TIME` range. 7. Supported only for unbounded `VARBINARY` (`BYTES`), because trimming or padding would corrupt the serialized bitmap data. +8. Supported iff every leaf type casts to `VARIANT` and every `MAP` key is a character string. Always fallible for `ARRAY` and `MAP`. Fallible for `ROW` and `STRUCTURED` iff a field is fallible, a field is a `VARIANT`, or the declared sizes of the fields can exceed the 16 MiB of a `VARIANT`. Also note that a cast of a `NULL` value will always return `NULL`, regardless of whether the function used is `CAST` or `TRY_CAST`. 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 ab0036b5a70..197da3ca43c 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 @@ -730,6 +730,23 @@ public final class LogicalTypeCasts { return allowExplicit && mapChildren.get(0).is(CHARACTER_STRING) && supportsCasting(sourceType, mapChildren.get(1), true); + } else if (targetRoot == VARIANT + && (sourceRoot == ARRAY + || sourceRoot == MAP + || sourceRoot == ROW + || sourceRoot == STRUCTURED_TYPE)) { + // An ARRAY becomes a variant array, and a MAP, ROW or STRUCTURED a variant object, so + // every element, value or field must cast to VARIANT. Object keys are strings, so a + // MAP needs a character string key, which is never converted from another type. + final List<LogicalType> sourceChildren = sourceType.getChildren(); + if (sourceRoot == MAP) { + return allowExplicit + && sourceChildren.get(0).is(CHARACTER_STRING) + && supportsCasting(sourceChildren.get(1), targetType, true); + } + return allowExplicit + && sourceChildren.stream() + .allMatch(child -> supportsCasting(child, targetType, true)); } else if (sourceRoot == RAW && !targetType.is(BINARY_STRING) && !targetType.is(CHARACTER_STRING) 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 a238417654b..f43d3b3eb46 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 @@ -445,17 +445,75 @@ class LogicalTypeCastsTest { 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), + // a whole constructed value casts into a single VARIANT when every leaf does + Arguments.of(new ArrayType(new IntType()), new VariantType(), false, true), + Arguments.of(new ArrayType(new VariantType()), new VariantType(), false, true), + Arguments.of(new ArrayType(new NullType()), new VariantType(), false, true), Arguments.of( new MapType(VarCharType.STRING_TYPE, new IntType()), new VariantType(), false, - false), + true), + Arguments.of( + new MapType(new CharType(3), new VariantType()), + new VariantType(), + false, + true), Arguments.of( new RowType(List.of(new RowField("a", new IntType()))), new VariantType(), false, + true), + Arguments.of(new RowType(List.of()), new VariantType(), false, true), + Arguments.of( + new RowType( + List.of( + new RowField( + "a", + new ArrayType( + new MapType( + VarCharType.STRING_TYPE, + new RowType( + List.of( + new RowField( + "b", + new TimestampType( + 9))))))))), + new VariantType(), + false, + true), + Arguments.of( + StructuredType.newBuilder(ObjectIdentifier.of("cat", "db", "User")) + .attributes( + Arrays.asList( + new StructuredAttribute("f1", new TimestampType()), + new StructuredAttribute("f2", new IntType()))) + .build(), + new VariantType(), + false, + true), + // a map key is never converted to a string, and every leaf needs a VARIANT kind + Arguments.of( + new MapType(new IntType(), new IntType()), new VariantType(), false, false), + Arguments.of( + new ArrayType( + new YearMonthIntervalType( + YearMonthIntervalType.YearMonthResolution.MONTH)), + new VariantType(), + false, + false), + Arguments.of( + new MapType(VarCharType.STRING_TYPE, new ZonedTimestampType()), + new VariantType(), + false, + false), + Arguments.of( + new RowType( + List.of( + new RowField( + "a", new MultisetType(VarCharType.STRING_TYPE)))), + new VariantType(), + false, false)); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java index 1a75db9dc7b..a14544853a1 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java @@ -24,17 +24,25 @@ import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.types.variant.BinaryVariantUtil; import org.apache.flink.types.variant.Variant; +import java.util.ArrayDeque; +import java.util.Deque; + import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.box; import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall; /** * Cast rule to {@link LogicalTypeRoot#VARIANT} from every type that {@link LogicalTypeCasts} - * allows. + * allows: a primitive type, or an {@link LogicalTypeRoot#ARRAY}, {@link LogicalTypeRoot#MAP}, + * {@link LogicalTypeRoot#ROW} or {@link LogicalTypeRoot#STRUCTURED_TYPE} whose leaves all cast to + * {@code VARIANT} and whose {@code MAP} keys are character strings. * * <p>The generated code calls a {@link ToVariantConverter} created once for the input type, which - * describes how a value is stored. See {@link #canFail} for when a cast can fail. + * describes how a value is stored. A failure anywhere in the value fails the whole cast, so {@code + * TRY_CAST} returns {@code NULL} rather than a partial result. See {@link #canFail} for when a cast + * can fail. */ class ToVariantCastRule extends AbstractNullAwareCodeGeneratorCastRule<Object, Variant> { @@ -43,6 +51,12 @@ class ToVariantCastRule extends AbstractNullAwareCodeGeneratorCastRule<Object, V /** 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; + /** The largest fixed-size value: a 16-byte decimal with its header and scale. */ + private static final long MAX_FIXED_SIZE = 18; + + /** A header byte and a 4-byte size or length, the largest header a value has. */ + private static final long MAX_HEADER_SIZE = 1 + BinaryVariantUtil.U32_SIZE; + private ToVariantCastRule() { super(CastRulePredicate.builder().predicate(ToVariantCastRule::matches).build()); } @@ -57,44 +71,83 @@ class ToVariantCastRule extends AbstractNullAwareCodeGeneratorCastRule<Object, V /** * 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. Three kinds of types can fail: + * {@code NULL} for it instead of failing. A value can fail in four ways: * * <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 {@code TIME} holds the milliseconds of one day. Only a source or a function can * produce a value outside that range, but such a value fails. - * <li>A string or binary value over {@link ToVariantConverter#MAX_PAYLOAD_BYTES} does not fit - * into a {@code VARIANT}. Only a type whose declared length allows such a value can fail. + * <li>The value does not fit into the 16 MiB of a {@code VARIANT}. Any {@code ARRAY}, {@code + * MAP} or nested {@code VARIANT} can reach that size. A string or binary type can when + * its declared length allows more, and a {@code ROW} when the declared sizes of its + * fields add up to more. + * <li>A {@code MAP} with a {@code NULL} key fails. Every {@code MAP} can already fail on its + * size, so this makes no other type fallible. * </ul> * * <p>Every other type never fails. * - * <p>The length check trusts the declared length. Flink does not enforce that length on values + * <p>The size 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) { - LogicalType type = inputLogicalType; - while (type.is(LogicalTypeRoot.DISTINCT_TYPE)) { - type = ((DistinctType) type).getSourceType(); + // Without an ARRAY, MAP or VARIANT, the size of a value is bounded by the sum of its + // leaves and object headers. + long maxSize = 0; + final Deque<LogicalType> pending = new ArrayDeque<>(); + pending.push(inputLogicalType); + while (!pending.isEmpty()) { + final LogicalType type = pending.pop(); + if (type.is(LogicalTypeRoot.DISTINCT_TYPE)) { + pending.push(((DistinctType) type).getSourceType()); + continue; + } + switch (type.getTypeRoot()) { + case ARRAY: + case MAP: + case VARIANT: + case TIME_WITHOUT_TIME_ZONE: + return true; + case ROW: + case STRUCTURED_TYPE: + LogicalTypeChecks.getFieldTypes(type).forEach(pending::push); + break; + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + if (LogicalTypeChecks.getPrecision(type) + > ToVariantConverter.TIMESTAMP_PRECISION) { + return true; + } + break; + default: + break; + } + maxSize += maxOwnSize(type); } + return maxSize > BinaryVariantUtil.SIZE_LIMIT; + } + + private static long maxOwnSize(LogicalType type) { switch (type.getTypeRoot()) { - case TIME_WITHOUT_TIME_ZONE: - return true; - case TIMESTAMP_WITHOUT_TIME_ZONE: - case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - return LogicalTypeChecks.getPrecision(type) > ToVariantConverter.TIMESTAMP_PRECISION; + case ROW: + case STRUCTURED_TYPE: + // Only the object header. Its fields are counted on their own. A 4-byte id per + // field, and a 4-byte offset per field plus one for the end. + return MAX_HEADER_SIZE + + BinaryVariantUtil.U32_SIZE + * (2L * LogicalTypeChecks.getFieldCount(type) + 1); case CHAR: case VARCHAR: - return (long) MAX_UTF8_BYTES_PER_CHAR * LogicalTypeChecks.getLength(type) - > ToVariantConverter.MAX_PAYLOAD_BYTES; + return MAX_HEADER_SIZE + + (long) MAX_UTF8_BYTES_PER_CHAR * LogicalTypeChecks.getLength(type); case BINARY: case VARBINARY: - return LogicalTypeChecks.getLength(type) > ToVariantConverter.MAX_PAYLOAD_BYTES; + return MAX_HEADER_SIZE + LogicalTypeChecks.getLength(type); default: - return false; + return MAX_FIXED_SIZE; } } 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 c3d2d4c2353..714df428b58 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 @@ -86,7 +86,9 @@ import static org.apache.flink.table.api.DataTypes.VARCHAR; import static org.apache.flink.table.api.DataTypes.VARIANT; import static org.apache.flink.table.api.DataTypes.YEAR; import static org.apache.flink.table.api.Expressions.$; +import static org.apache.flink.table.api.Expressions.array; import static org.apache.flink.table.api.Expressions.lit; +import static org.apache.flink.table.api.Expressions.row; import static org.apache.flink.util.CollectionUtil.entry; import static org.apache.flink.util.CollectionUtil.map; import static org.assertj.core.api.Assertions.assertThat; @@ -165,6 +167,7 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { specs.addAll(variantRowCasts()); specs.addAll(variantMapCasts()); specs.addAll(primitiveToVariantCasts()); + specs.addAll(constructedToVariantCasts()); return specs; } @@ -711,15 +714,172 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { $("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())); + } + + private static List<TestSetSpec> constructedToVariantCasts() { + final DataType person = ROW(FIELD("id", BIGINT()), FIELD("name", STRING())); + return List.of( + // A VARIANT object sorts its keys, so CAST to STRING shows the fields by name. + TestSetSpec.forExpression("Cast a constructed type to VARIANT and back") + .onFieldsWithData( + new Integer[] {1, null}, + Row.of("ada", 7L, new String[] {"x", null}), + map(entry("a", 1))) + .andDataTypes( + ARRAY(INT()), + ROW( + FIELD("name", STRING()), + FIELD("id", BIGINT()), + FIELD("tags", ARRAY(STRING()))), + MAP(STRING(), INT())) + .testResult( + $("f0").cast(VARIANT()).cast(ARRAY(INT())), + "CAST(CAST(f0 AS VARIANT) AS ARRAY<INT>)", + new Integer[] {1, null}, + ARRAY(INT())) + .testResult( + $("f1").cast(VARIANT()) + .cast( + ROW( + FIELD("name", STRING()), + FIELD("id", BIGINT()), + FIELD("tags", ARRAY(STRING())))), + "CAST(CAST(f1 AS VARIANT) AS ROW<name STRING, id BIGINT, tags ARRAY<STRING>>)", + Row.of("ada", 7L, new String[] {"x", null}), + ROW( + FIELD("name", STRING()), + FIELD("id", BIGINT()), + FIELD("tags", ARRAY(STRING())))) + .testResult( + $("f2").cast(VARIANT()).cast(MAP(STRING(), INT())), + "CAST(CAST(f2 AS VARIANT) AS MAP<STRING, INT>)", + map(entry("a", 1)), + MAP(STRING(), INT())) + .testResult( + $("f1").cast(VARIANT()).cast(STRING()), + "CAST(CAST(f1 AS VARIANT) AS STRING)", + "{id=7, name=ada, tags=[x, NULL]}", + STRING()) + // a field can be read back by its key + .testSqlResult("CAST(CAST(f1 AS VARIANT)['id'] AS BIGINT)", 7L, BIGINT()), + // a ROW element becomes one VARIANT, while a NULL element stays SQL NULL. Calcite + // turns a plain SQL CAST from an ARRAY of ROW to an ARRAY of another type into a + // MULTISET, so SQL uses TRY_CAST and the Table API is not tested as SQL. + TestSetSpec.forExpression("Cast ROW elements to VARIANT elements and back") + .onFieldsWithData( + new Row[] {Row.of(7L, "ada"), null}, + map(entry("a", Row.of(7L, "ada")))) + .andDataTypes(ARRAY(person), MAP(STRING(), person)) + .testTableApiResult( + $("f0").cast(ARRAY(VARIANT())).cast(ARRAY(person)), + new Row[] {Row.of(7L, "ada"), null}, + ARRAY(person)) + .testSqlResult( + "CAST(TRY_CAST(f0 AS ARRAY<VARIANT>) AS ARRAY<ROW<id BIGINT, name STRING>>)", + new Row[] {Row.of(7L, "ada"), null}, + ARRAY(person)) + .testTableApiResult( + $("f0").cast(ARRAY(VARIANT())).cast(ARRAY(STRING())), + new String[] {"{id=7, name=ada}", null}, + ARRAY(STRING())) + .testSqlResult( + "CAST(TRY_CAST(f0 AS ARRAY<VARIANT>) AS ARRAY<STRING>)", + new String[] {"{id=7, name=ada}", null}, + ARRAY(STRING())) + .testResult( + $("f1").cast(MAP(STRING(), VARIANT())).cast(MAP(STRING(), person)), + "CAST(CAST(f1 AS MAP<STRING, VARIANT>) AS MAP<STRING, ROW<id BIGINT, name STRING>>)", + map(entry("a", Row.of(7L, "ada"))), + MAP(STRING(), person)), + // A ROW is keyed by its field names. The SQL ROW constructor names them EXPR$0, + // EXPR$1 and so on, and a CAST to a named ROW renames them. + TestSetSpec.forExpression("Cast a ROW constructor to VARIANT") + .onFieldsWithData(1, "a") + .andDataTypes(INT(), STRING()) + .testSqlResult( + "CAST(CAST(ROW(f0, f1) AS VARIANT) AS STRING)", + "{EXPR$0=1, EXPR$1=a}", + STRING()) + .testSqlResult( + "CAST(CAST(CAST(ROW(f0, f1) AS ROW<id INT, name STRING>) AS VARIANT) AS STRING)", + "{id=1, name=a}", + STRING()) + .testTableApiResult( + row($("f0"), $("f1")).cast(VARIANT()).cast(STRING()), + "{f0=1, f1=a}", + STRING()) + .testTableApiResult( + row($("f0").as("id"), $("f1").as("name")) + .cast(VARIANT()) + .cast(STRING()), + "{id=1, name=a}", + STRING()) + // a STRUCTURED value is keyed by its attribute names + .testSqlResult( + "CAST(CAST(OBJECT_OF('com.example.Person', 'id', f0, 'name', f1) AS VARIANT) AS STRING)", + "{id=1, name=a}", STRING()), - // a whole constructed value does not cast into a single VARIANT yet - TestSetSpec.forExpression("Cast a constructed type to VARIANT") + TestSetSpec.forExpression("Cast a constructed literal to VARIANT") .onFieldsWithData(0) + // a cast from VARIANT to STRING is nullable, since a variant null becomes + // a SQL NULL + .testResult( + array(1, 2).cast(VARIANT()).cast(STRING()), + "CAST(CAST(ARRAY[1, 2] AS VARIANT) AS STRING)", + "[1, 2]", + STRING()) + .testSqlResult( + "CAST(CAST(MAP['a', 1, 'b', 2] AS VARIANT) AS STRING)", + "{a=1, b=2}", + STRING()) + // a variant object sorts its keys, for a MAP as for a ROW + .testSqlResult( + "CAST(CAST(MAP['b', 2, 'a', 1] AS VARIANT) AS STRING)", + "{a=1, b=2}", + STRING()), + // a nested VARIANT is embedded as is, including a field of another VARIANT + TestSetSpec.forExpression("Cast a constructed type with a VARIANT to VARIANT") + .onFieldsWithData("{\"a\":[1,2],\"b\":\"x\"}") + .andDataTypes(STRING()) + .testSqlResult( + "CAST(CAST(ARRAY[PARSE_JSON(f0)['a'], PARSE_JSON(f0)['b']] AS VARIANT) AS STRING)", + "[[1, 2], x]", + STRING()) + .testSqlResult( + "CAST(CAST(CAST(ROW(PARSE_JSON(f0)) AS ROW<v VARIANT>) AS VARIANT) AS STRING)", + "{v={a=[1, 2], b=x}}", + STRING()), + TestSetSpec.forExpression("Cast a failing constructed type to VARIANT") + .onFieldsWithData("x") + .andDataTypes(STRING()) + // a VARIANT object key cannot be NULL, and NULLIF makes the key NULL at + // runtime + .testSqlRuntimeError( + "CAST(MAP[NULLIF(f0, 'x'), 1] AS VARIANT)", + TableRuntimeException.class, + "A VARIANT object key cannot be NULL.") + .testSqlResult( + "TRY_CAST(MAP[NULLIF(f0, 'x'), 1] AS VARIANT)", null, VARIANT()) + // two strings of 9 MiB do not fit into the 16 MiB of a VARIANT together + .testSqlResult( + "TRY_CAST(ARRAY[REPEAT(f0, 9437184), REPEAT(f0, 9437184)] AS VARIANT)", + null, + VARIANT()), + // every leaf needs a VARIANT kind, and a map key is never converted to a string + TestSetSpec.forExpression("Cast an unsupported constructed type to VARIANT") + .onFieldsWithData(map(entry(1, "a"))) + .andDataTypes(MAP(INT(), STRING())) .testSqlValidationError( - "CAST(ARRAY[1, 2] AS VARIANT)", + "CAST(f0 AS VARIANT)", "Cast function cannot convert value") + .testTableApiValidationError($("f0").cast(VARIANT()), "Unsupported cast") + .testSqlValidationError( + "CAST(ARRAY[INTERVAL '1' DAY] AS VARIANT)", "Cast function cannot convert value") - .testTableApiValidationError( - lit(new int[] {1, 2}).cast(VARIANT()), "Unsupported cast")); + // MULTISET has no variant counterpart + .testSqlValidationError( + "CAST(MULTISET[1, 1] AS VARIANT)", + "Cast function cannot convert value")); } private static List<TestSetSpec> variantArrayCasts() { 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 788d849c50a..e63cd7d0eb5 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 @@ -230,6 +230,53 @@ class CastRuleProviderTest { .isTrue(); } + @Test + void testResolveConstructedToVariant() { + assertThat(CastRuleProvider.resolve(ARRAY(INT()).getLogicalType(), VARIANT)) + .isSameAs(ToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(MAP(STRING(), INT()).getLogicalType(), VARIANT)) + .isSameAs(ToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(ROW, VARIANT)).isSameAs(ToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(STRUCTURED, VARIANT)) + .isSameAs(ToVariantCastRule.INSTANCE); + + // every leaf needs a VARIANT kind, and a map key is never converted to a string + assertThat(CastRuleProvider.exists(ARRAY(INTERVAL(MONTH())).getLogicalType(), VARIANT)) + .isFalse(); + assertThat(CastRuleProvider.exists(MAP(INT(), STRING()).getLogicalType(), VARIANT)) + .isFalse(); + assertThat(CastRuleProvider.exists(MULTISET(STRING()).getLogicalType(), VARIANT)).isFalse(); + + // any ARRAY, MAP or nested VARIANT can exceed 16 MiB + assertThat(CastRuleProvider.canFail(ARRAY(INT()).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(MAP(STRING(), INT()).getLogicalType(), VARIANT)) + .isTrue(); + assertThat(CastRuleProvider.canFail(ROW(FIELD("v", VARIANT())).getLogicalType(), VARIANT)) + .isTrue(); + + // a ROW fails when a leaf can, or when the declared sizes of its fields add up past 16 MiB + assertThat(CastRuleProvider.canFail(ROW, VARIANT)).isFalse(); + assertThat(CastRuleProvider.canFail(STRUCTURED, VARIANT)).isFalse(); + assertThat( + CastRuleProvider.canFail( + ROW(FIELD("a", TIMESTAMP(9))).getLogicalType(), VARIANT)) + .isTrue(); + assertThat(CastRuleProvider.canFail(ROW(FIELD("a", TIME())).getLogicalType(), VARIANT)) + .isTrue(); + assertThat(CastRuleProvider.canFail(ROW(FIELD("a", STRING())).getLogicalType(), VARIANT)) + .isTrue(); + assertThat( + CastRuleProvider.canFail( + ROW(FIELD("a", VARCHAR(4_000_000))).getLogicalType(), VARIANT)) + .isFalse(); + assertThat( + CastRuleProvider.canFail( + ROW(FIELD("a", VARCHAR(4_000_000)), FIELD("b", VARCHAR(4_000_000))) + .getLogicalType(), + VARIANT)) + .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 abc7ec972e2..49e9f6ccace 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 @@ -2097,22 +2097,63 @@ class CastRulesTest { // 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 + // a constructed type casts element by element, and a NULL element stays SQL NULL. + // A constructed element becomes one VARIANT, like a cast of the element alone. CastTestSpecBuilder.testCastTo(ARRAY(VARIANT())) .fromCase( ARRAY(INT()), new GenericArrayData(new Integer[] {1, null}), - new GenericArrayData(new Object[] {VARIANT_BUILDER.of(1), null})), + new GenericArrayData(new Object[] {VARIANT_BUILDER.of(1), null})) + .fromCase( + ARRAY(ROW(FIELD("id", BIGINT()), FIELD("name", STRING()))), + new GenericArrayData( + new Object[] { + GenericRowData.of(7L, fromString("ada")), null + }), + new GenericArrayData( + new Object[] { + VARIANT_BUILDER + .object() + .add("id", VARIANT_BUILDER.of(7L)) + .add("name", VARIANT_BUILDER.of("ada")) + .build(), + 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)), + GenericRowData.of(VARIANT_BUILDER.of(7), null)) + .fromCase( + ROW( + FIELD("a", ROW(FIELD("id", BIGINT()))), + FIELD("b", ARRAY(INT()))), + GenericRowData.of( + GenericRowData.of(7L), new GenericArrayData(new int[] {1})), + GenericRowData.of( + VARIANT_BUILDER + .object() + .add("id", VARIANT_BUILDER.of(7L)) + .build(), + VARIANT_BUILDER + .array() + .add(VARIANT_BUILDER.of(1)) + .build())), CastTestSpecBuilder.testCastTo(MAP(STRING(), VARIANT())) .fromCase( MAP(STRING(), INT()), mapData(entry(fromString("a"), 1)), - mapData(entry(fromString("a"), VARIANT_BUILDER.of(1)))), + mapData(entry(fromString("a"), VARIANT_BUILDER.of(1)))) + .fromCase( + MAP(STRING(), ROW(FIELD("id", BIGINT()))), + mapData(entry(fromString("a"), GenericRowData.of(7L))), + mapData( + entry( + fromString("a"), + VARIANT_BUILDER + .object() + .add("id", VARIANT_BUILDER.of(7L)) + .build()))), CastTestSpecBuilder.testCastTo(MAP(VARIANT(), STRING())) .fromCase( MAP(INT(), STRING()), @@ -2123,6 +2164,44 @@ class CastRulesTest { MULTISET(INT()), mapData(entry(1, 2)), mapData(entry(VARIANT_BUILDER.of(1), 2))), + // a whole constructed value becomes one VARIANT, and a NULL element or field + // becomes a variant null + CastTestSpecBuilder.testCastTo(VARIANT()) + .fromCase( + ARRAY(INT()), + new GenericArrayData(new Integer[] {1, null}), + VARIANT_BUILDER + .array() + .add(VARIANT_BUILDER.of(1)) + .add(VARIANT_BUILDER.ofNull()) + .build()) + .fromCase( + MAP(STRING(), INT()), + mapData(entry(fromString("a"), 1)), + VARIANT_BUILDER.object().add("a", VARIANT_BUILDER.of(1)).build()) + .fromCase( + ROW(FIELD("id", BIGINT()), FIELD("tags", ARRAY(STRING()))), + GenericRowData.of( + 7L, new GenericArrayData(new Object[] {fromString("x")})), + VARIANT_BUILDER + .object() + .add("id", VARIANT_BUILDER.of(7L)) + .add( + "tags", + VARIANT_BUILDER + .array() + .add(VARIANT_BUILDER.of("x")) + .build()) + .build()) + .fromCase( + ROW(FIELD("v", VARIANT())), + GenericRowData.of(VARIANT_BUILDER.ofNull()), + VARIANT_BUILDER.object().add("v", VARIANT_BUILDER.ofNull()).build()) + .fail( + MAP(STRING(), INT()), + mapData(entry(null, 1)), + TableRuntimeException.class, + "A VARIANT object key cannot be NULL."), CastTestSpecBuilder.testCastTo(BOOLEAN()) .fromCase(VARIANT(), VARIANT_BUILDER.of(true), true) .fromCase(VARIANT(), VARIANT_BUILDER.of(false), false)
