diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 1cbc6f8bae63c..6ad9cd02c7a78 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1515,6 +1515,12 @@ without requiring upfront schema definition. For example, if a new field is adde can be directly incorporated into the `VARIANT` data without modifying the table schema. This is particularly useful in dynamic environments where schemas may evolve over time. +A primitive-valued `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. Numeric +targets are lenient: a variant holding any numeric value casts to any numeric type, so a JSON integer +such as `PARSE_JSON('42')` casts to `INT` or `BIGINT`. Other targets require the stored value to be of +the matching kind. When the value cannot be converted, `CAST` fails the job and `TRY_CAST` returns +`NULL`. Use the `JSON_STRING` function to obtain the JSON string representation of a `VARIANT`. + **Declaration** {{< tabs "25c30432-8460-441d-a036-9416d8202882" >}} @@ -1729,7 +1735,7 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 | `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 | | `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 | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | +| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | N | N | N | 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 | 备注： diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index a849ee5dac8de..45ace31e36508 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1523,6 +1523,12 @@ without requiring upfront schema definition. For example, if a new field is adde can be directly incorporated into the `VARIANT` data without modifying the table schema. This is particularly useful in dynamic environments where schemas may evolve over time. +A primitive-valued `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. Numeric +targets are lenient: a variant holding any numeric value casts to any numeric type, so a JSON integer +such as `PARSE_JSON('42')` casts to `INT` or `BIGINT`. Other targets require the stored value to be of +the matching kind. When the value cannot be converted, `CAST` fails the job and `TRY_CAST` returns +`NULL`. Use the `JSON_STRING` function to obtain the JSON string representation of a `VARIANT`. + **Declaration** {{< tabs "25c30432-8460-441d-a036-9416d8202882" >}} @@ -1738,7 +1744,7 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | `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 | | `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 | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | +| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | N | N | N | 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 | Notes: diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java index fbd4bf188d751..3bd66e01cd5ec 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java @@ -35,6 +35,7 @@ import java.util.List; import java.util.Optional; +import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.getUnsupportedCastHint; import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsExplicitCast; /** @@ -69,6 +70,15 @@ public Optional> inferInputTypes( return Optional.of(argumentDataTypes); } if (!supportsExplicitCast(fromType, toType)) { + final Optional hint = getUnsupportedCastHint(fromType, toType); + if (hint.isPresent()) { + return callContext.fail( + throwOnFailure, + "Unsupported cast from '%s' to '%s'. %s", + fromType, + toType, + hint.get()); + } return callContext.fail( throwOnFailure, "Unsupported cast from '%s' to '%s'.", fromType, toType); } 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 a0afca2caa09c..58f474ec0eb70 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 @@ -37,6 +37,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.function.BiFunction; import java.util.function.BiPredicate; @@ -79,6 +80,7 @@ import static org.apache.flink.table.types.logical.LogicalTypeRoot.TINYINT; import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARBINARY; import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARCHAR; +import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARIANT; import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getDayPrecision; import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getFractionalPrecision; import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getLength; @@ -646,6 +648,9 @@ private static boolean supportsCasting( // BITMAP can only be cast to BYTES (unbounded VARBINARY), because trimming or padding // would corrupt the serialized bitmap data. return allowExplicit && getLength(targetType) == VarBinaryType.MAX_LENGTH; + } else if (sourceRoot == VARIANT) { + // a VARIANT can only be explicitly cast to a supported scalar type + return allowExplicit && supportsVariantToScalarCast(targetType); } if (implicitCastingRules.get(targetRoot).contains(sourceRoot)) { @@ -726,6 +731,45 @@ private static boolean supportsConstructedCasting( return false; } + private static boolean supportsVariantToScalarCast(LogicalType targetType) { + switch (targetType.getTypeRoot()) { + case BOOLEAN: + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + case BINARY: + case VARBINARY: + case DATE: + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return true; + default: + // TIME has no counterpart in the Variant type model. CHARACTER_STRING is handled by + // the display-oriented VariantToStringCastRule and is intentionally not offered as + // a + // user-facing cast here. + return false; + } + } + + /** + * Returns a hint pointing to the function that performs a conceptually related operation when + * an explicit cast is unsupported, or empty when no specific hint applies. + */ + public static Optional getUnsupportedCastHint( + LogicalType sourceType, LogicalType targetType) { + if (sourceType.is(VARIANT) && targetType.is(CHARACTER_STRING)) { + return Optional.of( + "Use the JSON_STRING function to convert a VARIANT to its JSON string " + + "representation."); + } + return Optional.empty(); + } + private static CastingRuleBuilder castTo(LogicalTypeRoot targetType) { return new CastingRuleBuilder(targetType); } 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 3d0dec239974f..6fb3574b1f59f 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 @@ -34,6 +34,7 @@ import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.NullType; import org.apache.flink.table.types.logical.RawType; import org.apache.flink.table.types.logical.RowType; @@ -46,6 +47,7 @@ import org.apache.flink.table.types.logical.TinyIntType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.table.types.logical.VariantType; import org.apache.flink.table.types.logical.YearMonthIntervalType; import org.apache.flink.table.types.logical.ZonedTimestampType; import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; @@ -57,6 +59,7 @@ import org.junit.jupiter.params.provider.MethodSource; import java.util.Arrays; +import java.util.List; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -262,7 +265,37 @@ private static Stream testData() { new RawType<>(Integer.class, IntSerializer.INSTANCE), VarCharType.STRING_TYPE, false, - true)); + true), + + // variant to scalar is explicit only + Arguments.of(new VariantType(), new BooleanType(), false, true), + Arguments.of(new VariantType(), new TinyIntType(), false, true), + Arguments.of(new VariantType(), new IntType(), false, true), + Arguments.of(new VariantType(), new BigIntType(), false, true), + Arguments.of(new VariantType(), new DoubleType(), false, true), + Arguments.of(new VariantType(), new DecimalType(10, 2), false, true), + Arguments.of(new VariantType(), new DateType(), false, true), + Arguments.of(new VariantType(), new TimestampType(), false, true), + Arguments.of(new VariantType(), new TimestampType(3), false, true), + Arguments.of(new VariantType(), new LocalZonedTimestampType(), false, true), + Arguments.of(new VariantType(), new LocalZonedTimestampType(9), false, true), + Arguments.of( + new VariantType(), + new VarBinaryType(VarBinaryType.MAX_LENGTH), + false, + true), + // variant identity cast is implicit + Arguments.of(new VariantType(), new VariantType(), true, true), + // TIME, character strings and constructed targets are not castable from variant + Arguments.of(new VariantType(), new TimeType(), false, false), + Arguments.of(new VariantType(), VarCharType.STRING_TYPE, false, false), + Arguments.of(new VariantType(), new ArrayType(new IntType()), false, false), + Arguments.of(new VariantType(), new RowType(List.of()), false, false), + Arguments.of( + new VariantType(), + new MapType(new IntType(), new CharType()), + 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/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 b18b44e126e17..133542ec6f58e 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 @@ -97,6 +97,7 @@ public class CastRuleProvider { .addRule(RowToRowCastRule.INSTANCE) // Variant rules .addRule(VariantToStringCastRule.INSTANCE) + .addRule(VariantToPrimitiveCastRule.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/VariantToPrimitiveCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java new file mode 100644 index 0000000000000..8a278f41498b5 --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java @@ -0,0 +1,237 @@ +/* + * 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.data.DecimalData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.planner.functions.casting.CastRuleUtils.CodeWriter; +import org.apache.flink.table.types.logical.DecimalType; +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 java.math.BigDecimal; +import java.util.Arrays; + +import static org.apache.flink.table.planner.codegen.CodeGenUtils.className; +import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.arrayLength; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.cast; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.ternaryOperator; + +/** + * {@link LogicalTypeRoot#VARIANT} to primitive type cast rule. + * + *

Numeric targets are lenient and follow regular numeric cast semantics; other targets require + * the stored value to match the target kind. On a mismatch {@code CAST} fails and {@code TRY_CAST} + * returns {@code null}. + * + *

{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule}; {@code TIME} has no + * variant counterpart and is unsupported. + */ +class VariantToPrimitiveCastRule extends AbstractNullAwareCodeGeneratorCastRule { + + static final VariantToPrimitiveCastRule INSTANCE = new VariantToPrimitiveCastRule(); + + private VariantToPrimitiveCastRule() { + super( + CastRulePredicate.builder() + .predicate( + (input, target) -> + input.is(LogicalTypeRoot.VARIANT) + && isSupportedTarget(target)) + .build()); + } + + private static boolean isSupportedTarget(LogicalType targetType) { + switch (targetType.getTypeRoot()) { + case BOOLEAN: + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + case BINARY: + case VARBINARY: + case DATE: + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return true; + default: + return false; + } + } + + @Override + public boolean canFail(LogicalType inputLogicalType, LogicalType targetLogicalType) { + return true; + } + + /** + * Treats a variant that stores a JSON {@code null} as a {@code NULL} input, so it casts to SQL + * {@code NULL} instead of failing in the type-specific accessor. Only applied for a nullable + * target: a {@code NOT NULL} result cannot carry {@code NULL}, so a null-valued variant then + * fails as a regular type mismatch. + */ + @Override + public CastCodeBlock generateCodeBlock( + CodeGeneratorCastRule.Context context, + String inputTerm, + String inputIsNullTerm, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + if (!targetLogicalType.isNullable()) { + return super.generateCodeBlock( + context, inputTerm, inputIsNullTerm, inputLogicalType, targetLogicalType); + } + final String isNullTerm = + "(" + inputIsNullTerm + " || " + methodCall(inputTerm, "isNull") + ")"; + return super.generateCodeBlock( + context, inputTerm, isNullTerm, inputLogicalType, targetLogicalType); + } + + @Override + protected String generateCodeBlockInternal( + CodeGeneratorCastRule.Context context, + String inputTerm, + String returnVariable, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + final CodeWriter writer = new CastRuleUtils.CodeWriter(); + switch (targetLogicalType.getTypeRoot()) { + case BOOLEAN: + writer.assignStmt(returnVariable, methodCall(inputTerm, "getBoolean")); + break; + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + writer.assignStmt(returnVariable, numericExpression(inputTerm, targetLogicalType)); + break; + case BINARY: + case VARBINARY: + generateToBytes(context, inputTerm, returnVariable, targetLogicalType, writer); + break; + case DATE: + writer.assignStmt( + returnVariable, + cast("int", methodCall(methodCall(inputTerm, "getDate"), "toEpochDay"))); + break; + case TIMESTAMP_WITHOUT_TIME_ZONE: + writer.assignStmt( + returnVariable, + staticCall( + TimestampData.class, + "fromLocalDateTime", + methodCall(inputTerm, "getDateTime"))); + break; + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + writer.assignStmt( + returnVariable, + staticCall( + TimestampData.class, + "fromInstant", + methodCall(inputTerm, "getInstant"))); + break; + default: + throw new IllegalArgumentException( + "Unsupported target type for casting from VARIANT: " + targetLogicalType); + } + return writer.toString(); + } + + /** + * Converts a numeric variant to the numeric {@code target} via the matching {@link Number} + * accessor, mirroring regular numeric cast semantics. A non-numeric variant raises {@link + * ClassCastException}, failing {@code CAST} and yielding {@code null} for {@code TRY_CAST}. + */ + private static String numericExpression(String inputTerm, LogicalType target) { + final String number = cast(className(Number.class), methodCall(inputTerm, "get")); + if (!target.is(LogicalTypeRoot.DECIMAL)) { + return methodCall(number, numberAccessor(target)); + } + final DecimalType decimalType = (DecimalType) target; + return staticCall( + DecimalData.class, + "fromBigDecimal", + constructorCall(BigDecimal.class, methodCall(number, "toString")), + decimalType.getPrecision(), + decimalType.getScale()); + } + + private static String numberAccessor(LogicalType target) { + switch (target.getTypeRoot()) { + case TINYINT: + return "byteValue"; + case SMALLINT: + return "shortValue"; + case INTEGER: + return "intValue"; + case BIGINT: + return "longValue"; + case FLOAT: + return "floatValue"; + case DOUBLE: + return "doubleValue"; + default: + throw new IllegalArgumentException( + "Unsupported numeric target for casting from VARIANT: " + target); + } + } + + private static void generateToBytes( + CodeGeneratorCastRule.Context context, + String inputTerm, + String returnVariable, + LogicalType targetLogicalType, + CodeWriter writer) { + final int targetLength = LogicalTypeChecks.getLength(targetLogicalType); + // Read the bytes once to avoid decoding the variant twice. + final String bytesTerm = newName(context.getCodeGeneratorContext(), "variantBytes"); + writer.declStmt("byte[]", bytesTerm, methodCall(inputTerm, "getBytes")); + if (BinaryToBinaryCastRule.couldPad(targetLogicalType, targetLength)) { + // BINARY(n): pad or trim to the exact target length. + writer.assignStmt( + returnVariable, + ternaryOperator( + arrayLength(bytesTerm) + " == " + targetLength, + bytesTerm, + staticCall(Arrays.class, "copyOf", bytesTerm, targetLength))); + } else if (BinaryToBinaryCastRule.couldTrim(targetLength)) { + // VARBINARY(n): trim only when longer than the target length. + writer.assignStmt( + returnVariable, + ternaryOperator( + arrayLength(bytesTerm) + " <= " + targetLength, + bytesTerm, + staticCall(Arrays.class, "copyOf", bytesTerm, targetLength))); + } else { + writer.assignStmt(returnVariable, bytesTerm); + } + } +} 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 5420ce1987f10..cb41ad083c741 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 @@ -78,6 +78,7 @@ import static org.apache.flink.table.api.DataTypes.VARCHAR; 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.call; import static org.apache.flink.util.CollectionUtil.entry; import static org.apache.flink.util.CollectionUtil.map; import static org.assertj.core.api.Assertions.assertThat; @@ -132,9 +133,111 @@ Stream getTestSetSpecs() { specs.addAll(numericBounds()); specs.addAll(constructedTypes()); specs.addAll(bitmapCasts()); + specs.addAll(variantCasts()); return specs.stream(); } + private static List variantCasts() { + // A variant is produced with PARSE_JSON so that the source column stays a STRING literal; + // there is no VARIANT literal to feed a source column directly. A JSON integer is stored in + // the smallest integer type that fits (42 -> TINYINT), and numeric casts are lenient, so it + // still widens to INT. + return List.of( + TestSetSpec.forExpression("Cast a VARIANT produced by PARSE_JSON to a primitive") + .onFieldsWithData("unused") + .andDataTypes(STRING()) + .testResult( + call("PARSE_JSON", "42").cast(INT()), + "CAST(PARSE_JSON('42') AS INT)", + 42, + INT().notNull()) + // Integer overflow wraps around (Java narrowing), like a regular numeric + // cast. + .testResult( + call("PARSE_JSON", "40000").cast(SMALLINT()), + "CAST(PARSE_JSON('40000') AS SMALLINT)", + (short) -25536, + SMALLINT().notNull()) + .testResult( + call("PARSE_JSON", "128").cast(TINYINT()), + "CAST(PARSE_JSON('128') AS TINYINT)", + (byte) -128, + TINYINT().notNull()) + .testResult( + call("PARSE_JSON", "2147483648").cast(INT()), + "CAST(PARSE_JSON('2147483648') AS INT)", + -2147483648, + INT().notNull()) + .testResult( + call("PARSE_JSON", "9223372036854775808").cast(BIGINT()), + "CAST(PARSE_JSON('9223372036854775808') AS BIGINT)", + -9223372036854775808L, + BIGINT().notNull()) + // An out-of-range floating point value saturates when cast to an integer, + // and a fractional value is truncated toward zero. + .testResult( + call("PARSE_JSON", "1e20").cast(INT()), + "CAST(PARSE_JSON('1e20') AS INT)", + 2147483647, + INT().notNull()) + .testResult( + call("PARSE_JSON", "3.9").cast(INT()), + "CAST(PARSE_JSON('3.9') AS INT)", + 3, + INT().notNull()) + // A value beyond the FLOAT range becomes infinity. + .testResult( + call("PARSE_JSON", "1e40").cast(FLOAT()), + "CAST(PARSE_JSON('1e40') AS FLOAT)", + Float.POSITIVE_INFINITY, + FLOAT().notNull()) + // DECIMAL overflow yields NULL instead of wrapping; TRY_CAST surfaces it. + .testResult( + call("PARSE_JSON", "123.456").tryCast(DECIMAL(4, 2)), + "TRY_CAST(PARSE_JSON('123.456') AS DECIMAL(4, 2))", + null, + DECIMAL(4, 2)) + .testResult( + call("PARSE_JSON", "42").cast(BIGINT()), + "CAST(PARSE_JSON('42') AS BIGINT)", + 42L, + BIGINT().notNull()) + .testResult( + call("PARSE_JSON", "true").cast(BOOLEAN()), + "CAST(PARSE_JSON('true') AS BOOLEAN)", + true, + BOOLEAN().notNull()) + // TRY_CAST of a value whose kind does not match the target returns NULL + .testResult( + call("PARSE_JSON", "\"foo\"").tryCast(INT()), + "TRY_CAST(PARSE_JSON('\"foo\"') AS INT)", + null, + INT()) + // A variant that stores a JSON null casts to SQL NULL when the target is + // nullable: a nullable variant source, or TRY_CAST which forces nullable. + .testResult( + call("TRY_PARSE_JSON", "null").cast(INT()), + "CAST(TRY_PARSE_JSON('null') AS INT)", + null, + INT()) + .testResult( + call("PARSE_JSON", "null").tryCast(BOOLEAN()), + "TRY_CAST(PARSE_JSON('null') AS BOOLEAN)", + null, + BOOLEAN()) + // A nullable variant with a concrete value still casts normally. + .testResult( + call("TRY_PARSE_JSON", "42").cast(INT()), + "CAST(TRY_PARSE_JSON('42') AS INT)", + 42, + INT()) + // Casting a VARIANT to a string points the user to JSON_STRING. + .testTableApiValidationError( + call("PARSE_JSON", "42").cast(STRING()), + "Use the JSON_STRING function to convert a VARIANT to its JSON " + + "string representation.")); + } + private static List allTypesBasic() { return Arrays.asList( CastTestSpecBuilder.testCastTo(BOOLEAN()) @@ -1269,7 +1372,7 @@ private static List bitmapCasts() { } private static List decimalCasts() { - return Collections.singletonList( + return List.of( CastTestSpecBuilder.testCastTo(DECIMAL(8, 4)) .fromCase(STRING(), null, null) // rounding 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 4e230667111bd..adc44c474be48 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 @@ -28,13 +28,20 @@ import org.junit.jupiter.api.Test; 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.DATE; +import static org.apache.flink.table.api.DataTypes.DECIMAL; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.INT; import static org.apache.flink.table.api.DataTypes.ROW; import static org.apache.flink.table.api.DataTypes.STRING; import static org.apache.flink.table.api.DataTypes.STRUCTURED; import static org.apache.flink.table.api.DataTypes.TIME; +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.VARIANT; import static org.apache.flink.table.types.logical.VarCharType.STRING_TYPE; import static org.assertj.core.api.Assertions.assertThat; @@ -48,6 +55,7 @@ class CastRuleProviderTest { .build(); private static final LogicalType INT = INT().getLogicalType(); private static final LogicalType TINYINT = TINYINT().getLogicalType(); + private static final LogicalType VARIANT = VARIANT().getLogicalType(); private static final LogicalType ROW = ROW(FIELD("a", INT()), FIELD("b", TINYINT().notNull())).getLogicalType(); private static final LogicalType STRUCTURED = @@ -108,4 +116,24 @@ void testCanFail() { assertThat(CastRuleProvider.canFail(inputType, ROW(INT(), STRING()).getLogicalType())) .isFalse(); } + + @Test + void testResolveVariantToPrimitive() { + assertThat(CastRuleProvider.resolve(VARIANT, INT)) + .isSameAs(VariantToPrimitiveCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(VARIANT, BOOLEAN().getLogicalType())) + .isSameAs(VariantToPrimitiveCastRule.INSTANCE); + assertThat(CastRuleProvider.exists(VARIANT, DECIMAL(10, 2).getLogicalType())).isTrue(); + assertThat(CastRuleProvider.exists(VARIANT, DATE().getLogicalType())).isTrue(); + assertThat(CastRuleProvider.exists(VARIANT, TIMESTAMP().getLogicalType())).isTrue(); + assertThat(CastRuleProvider.exists(VARIANT, TIMESTAMP_LTZ().getLogicalType())).isTrue(); + assertThat(CastRuleProvider.exists(VARIANT, BYTES().getLogicalType())).isTrue(); + assertThat(CastRuleProvider.canFail(VARIANT, INT)).isTrue(); + + // TIME has no variant counterpart and is not castable + assertThat(CastRuleProvider.exists(VARIANT, TIME().getLogicalType())).isFalse(); + // character strings keep going through the display-oriented rule + assertThat(CastRuleProvider.resolve(VARIANT, STRING_TYPE)) + .isSameAs(VariantToStringCastRule.INSTANCE); + } } 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 2aa6e198a98fa..8459a804163ea 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 @@ -41,6 +41,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.Variant; import org.junit.jupiter.api.DynamicTest; import org.junit.jupiter.api.TestFactory; @@ -50,6 +51,7 @@ import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; @@ -95,6 +97,7 @@ import static org.apache.flink.table.api.DataTypes.TINYINT; 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.api.DataTypes.YEAR; import static org.apache.flink.table.data.DecimalData.fromBigDecimal; import static org.apache.flink.table.data.StringData.fromString; @@ -1540,7 +1543,100 @@ Stream testCases() { CastTestSpecBuilder.testCastTo(BYTES()) .fromCase(BITMAP(), DEFAULT_BITMAP, DEFAULT_BITMAP.toBytes()) .fromCase(BITMAP(), Bitmap.empty(), Bitmap.empty().toBytes()) - .fromCase(BITMAP(), null, null)); + .fromCase(BITMAP(), null, null), + // From VARIANT to primitive types. Numeric targets are lenient: a variant holding + // any numeric kind converts to the requested numeric type (widening and narrowing). + // Non-numeric targets are strict: the stored kind must match, otherwise the cast + // fails and TRY_CAST returns null. + CastTestSpecBuilder.testCastTo(BOOLEAN()) + .fromCase(VARIANT(), Variant.newBuilder().of(true), true) + .fromCase(VARIANT(), Variant.newBuilder().of(false), false) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(TINYINT()) + .fromCase(VARIANT(), Variant.newBuilder().of((byte) 42), (byte) 42) + .fromCase(VARIANT(), Variant.newBuilder().of(42), (byte) 42) + .fail(VARIANT(), Variant.newBuilder().of("x"), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(SMALLINT()) + .fromCase(VARIANT(), Variant.newBuilder().of((short) 42), (short) 42) + .fromCase(VARIANT(), Variant.newBuilder().of((byte) 42), (short) 42) + .fail( + VARIANT(), + Variant.newBuilder().of(true), + TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(INT()) + // widening: a JSON integer is stored in the smallest type but still casts + // up + .fromCase(VARIANT(), Variant.newBuilder().of((byte) 42), 42) + .fromCase(VARIANT(), Variant.newBuilder().of((short) 42), 42) + .fromCase(VARIANT(), Variant.newBuilder().of(42), 42) + .fromCase(VARIANT(), Variant.newBuilder().of(42L), 42) + // narrowing from a floating point or decimal value truncates + .fromCase(VARIANT(), Variant.newBuilder().of(3.9d), 3) + .fromCase(VARIANT(), Variant.newBuilder().of(new BigDecimal("7.2")), 7) + // a non-numeric variant cannot be cast to a number + .fail( + VARIANT(), + Variant.newBuilder().of("foo"), + TableRuntimeException.class) + .fail( + VARIANT(), + Variant.newBuilder().of(true), + TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(BIGINT()) + .fromCase(VARIANT(), Variant.newBuilder().of(42L), 42L) + .fromCase(VARIANT(), Variant.newBuilder().of(42), 42L) + .fail(VARIANT(), Variant.newBuilder().of("x"), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(FLOAT()) + .fromCase(VARIANT(), Variant.newBuilder().of(1.5f), 1.5f) + .fromCase(VARIANT(), Variant.newBuilder().of(1.5d), 1.5f) + .fromCase(VARIANT(), Variant.newBuilder().of(3), 3.0f) + .fail(VARIANT(), Variant.newBuilder().of("x"), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(DOUBLE()) + .fromCase(VARIANT(), Variant.newBuilder().of(1.5d), 1.5d) + .fromCase(VARIANT(), Variant.newBuilder().of(1.5f), 1.5d) + .fromCase(VARIANT(), Variant.newBuilder().of(3), 3.0d) + .fail(VARIANT(), Variant.newBuilder().of("x"), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(DECIMAL(5, 2)) + .fromCase(VARIANT(), null, null) + .fromCase( + VARIANT(), + Variant.newBuilder().of(new BigDecimal("123.45")), + DecimalData.fromBigDecimal(new BigDecimal("123.45"), 5, 2)) + .fromCase( + VARIANT(), + Variant.newBuilder().of(42), + DecimalData.fromBigDecimal(new BigDecimal("42"), 5, 2)) + .fail(VARIANT(), Variant.newBuilder().of("x"), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(BYTES()) + .fromCase(VARIANT(), null, null) + .fromCase( + VARIANT(), + Variant.newBuilder().of(new byte[] {1, 2, 3}), + new byte[] {1, 2, 3}) + .fail( + VARIANT(), + Variant.newBuilder().of("foo"), + TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(DATE()) + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalDate.of(2020, 1, 1)), + (int) LocalDate.of(2020, 1, 1).toEpochDay()) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(TIMESTAMP()) + .fromCase(VARIANT(), null, null) + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalDateTime.of(2020, 1, 1, 12, 0, 0)), + TimestampData.fromLocalDateTime( + LocalDateTime.of(2020, 1, 1, 12, 0, 0))) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class), + CastTestSpecBuilder.testCastTo(TIMESTAMP_LTZ()) + .fromCase( + VARIANT(), + Variant.newBuilder().of(Instant.ofEpochSecond(1_600_000_000L)), + TimestampData.fromInstant(Instant.ofEpochSecond(1_600_000_000L))) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class)); } @TestFactory