From d1f1a22d6910116b9294d939250eabb74c5b81e4 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 13 Jul 2026 13:28:59 +0200 Subject: [PATCH 1/5] [FLINK-37925][table] Support casting from VARIANT to primitive types Add CAST/TRY_CAST from a VARIANT to primitive scalar types. Numeric targets are lenient: a variant holding any numeric kind converts to the requested numeric type with regular numeric cast semantics, so a JSON integer stored in the smallest fitting type still widens (e.g. CAST(PARSE_JSON('42') AS INT)). Non-numeric targets require the stored value to be of the matching kind. On a type mismatch CAST fails the job and TRY_CAST returns NULL. Character strings and TIME are out of scope: strings are handled by the display-oriented VariantToStringCastRule and via JSON_STRING, and the variant type model has no TIME kind. Constructed types (ARRAY, ROW, STRUCTURED) are left to FLINK-37926. --- .../types/logical/utils/LogicalTypeCasts.java | 29 +++ .../functions/casting/CastRuleProvider.java | 1 + .../casting/VariantToPrimitiveCastRule.java | 214 ++++++++++++++++++ 3 files changed, 244 insertions(+) create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java 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..d079687a9aace 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 @@ -79,6 +79,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 +647,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 +730,31 @@ 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; + } + } + private static CastingRuleBuilder castTo(LogicalTypeRoot targetType) { return new CastingRuleBuilder(targetType); } 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..7d54c6514a50a --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java @@ -0,0 +1,214 @@ +/* + * 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; + } + + @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); + } + } +} From 6c422beb7b4ef26366edda2d9c40ebafa2ef6cc3 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 13 Jul 2026 13:28:59 +0200 Subject: [PATCH 2/5] [FLINK-37925][table] Add tests for casting from VARIANT to primitive types Cover cast validation (LogicalTypeCastsTest), rule resolution (CastRuleProviderTest), the generated cast executor including numeric widening/narrowing and type-mismatch failures (CastRulesTest), and an end-to-end CAST/TRY_CAST over PARSE_JSON (CastFunctionITCase). --- .../table/types/LogicalTypeCastsTest.java | 25 ++++- .../planner/functions/CastFunctionITCase.java | 34 +++++++ .../casting/CastRuleProviderTest.java | 28 ++++++ .../functions/casting/CastRulesTest.java | 98 ++++++++++++++++++- 4 files changed, 183 insertions(+), 2 deletions(-) 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..aab0d0be95943 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 @@ -46,6 +46,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; @@ -262,7 +263,29 @@ 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 LocalZonedTimestampType(), 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)); } @ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Implicit: {2}, Explicit: {3}]") 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..50cf8eefdbef8 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,42 @@ 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 Collections.singletonList( + 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()) + .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())); + } + private static List allTypesBasic() { return Arrays.asList( CastTestSpecBuilder.testCastTo(BOOLEAN()) 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 From cca9d90c42db56712eb36069eb8f88e9990c0d32 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 13 Jul 2026 13:28:59 +0200 Subject: [PATCH 3/5] [FLINK-37925][docs] Document casting from VARIANT to primitive types Mark VARIANT as an explicit cast source for scalar targets in the casting matrix and describe the lenient numeric behavior. Note that TIME and character-string targets are unsupported. --- docs/content.zh/docs/sql/reference/data-types.md | 8 +++++++- docs/content/docs/sql/reference/data-types.md | 8 +++++++- 2 files changed, 14 insertions(+), 2 deletions(-) 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: From 3bfbb4329fd0c00f23e9bc6ec4924ef4ff969607 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 13 Jul 2026 22:31:31 +0200 Subject: [PATCH 4/5] [FLINK-37925][table] Cast a null-valued VARIANT to NULL and hint JSON_STRING for VARIANT-to-string casts A variant that stores a JSON null now casts to SQL NULL instead of failing in the type-specific accessor, when the cast target is nullable (a nullable variant source, or TRY_CAST which forces the target nullable). A NOT NULL target still fails on a null-valued variant because the result cannot carry NULL; making VARIANT casts unconditionally nullable would require changes to Calcite return-type inference and is left out of scope. Casting a VARIANT to a character string now reports a helpful validation error from the Table API type inference that points to JSON_STRING. The SQL path keeps Calcite's generic cast error. --- .../strategies/CastInputTypeStrategy.java | 10 ++++++++ .../types/logical/utils/LogicalTypeCasts.java | 15 +++++++++++ .../casting/VariantToPrimitiveCastRule.java | 23 +++++++++++++++++ .../planner/functions/CastFunctionITCase.java | 25 ++++++++++++++++++- 4 files changed, 72 insertions(+), 1 deletion(-) 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 d079687a9aace..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; @@ -755,6 +756,20 @@ private static boolean supportsVariantToScalarCast(LogicalType targetType) { } } + /** + * 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-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 index 7d54c6514a50a..8a278f41498b5 100644 --- 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 @@ -89,6 +89,29 @@ public boolean canFail(LogicalType inputLogicalType, LogicalType targetLogicalTy 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, 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 50cf8eefdbef8..7356b4a65bf98 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 @@ -166,7 +166,30 @@ private static List variantCasts() { call("PARSE_JSON", "\"foo\"").tryCast(INT()), "TRY_CAST(PARSE_JSON('\"foo\"') AS INT)", null, - INT())); + 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() { From 0ebbdb13ccc412ac3919eac0b3a7e2e32a4319ea Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 13 Jul 2026 22:34:15 +0200 Subject: [PATCH 5/5] [FLINK-37925] Add more tests in LogicalTypeCastsTest --- .../table/types/LogicalTypeCastsTest.java | 12 ++++- .../planner/functions/CastFunctionITCase.java | 50 ++++++++++++++++++- 2 files changed, 59 insertions(+), 3 deletions(-) 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 aab0d0be95943..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; @@ -58,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; @@ -274,7 +276,9 @@ private static Stream testData() { 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), @@ -285,7 +289,13 @@ private static Stream testData() { // 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 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/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 7356b4a65bf98..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 @@ -142,7 +142,7 @@ private static List variantCasts() { // 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 Collections.singletonList( + return List.of( TestSetSpec.forExpression("Cast a VARIANT produced by PARSE_JSON to a primitive") .onFieldsWithData("unused") .andDataTypes(STRING()) @@ -151,6 +151,52 @@ private static List variantCasts() { "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)", @@ -1326,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