This is an automated email from the ASF dual-hosted git repository. RocMarshal pushed a commit to branch release-2.4 in repository https://gitbox.apache.org/repos/asf/flink.git
commit d545d196f74386e9da8f3dae60fddb46488e6a7f Author: Ramin Gharib <[email protected]> AuthorDate: Fri Nov 6 10:23:31 2026 +0100 [FLINK-41301][table] Throw TableRuntimeException for every failed cast from VARIANT The BOOLEAN reader of FromVariantConverter called Variant#getBoolean, which throws a VariantTypeException with the getter's message, such as "Expected type BOOLEAN but got STRING". It now checks the kind and throws the unsupportedKind TableRuntimeException, like the other readers. The DATE reader already does this since FLINK-41281, and the binary readers since FLINK-40319. A malformed variant, or one with a type this version does not know, still made a reader throw a VariantTypeException. FromVariantConverter#convert now wraps it in a TableRuntimeException and keeps it as the cause. A cast to a nullable target first checks for a variant null. That check reads the type outside convert and threw the same VariantTypeException. It now calls FromVariantConverter#isVariantNull, which wraps it the same way. The VARIANT_GET part of the master commit is left out, because release-2.4 has no VARIANT_GET. Generated-by: Claude Code 2.1.319 (Claude Opus 5.5) (cherry picked from commit de2cb88cb0c5a1c076364e02dd03a19b6d890b91) --- .../table/data/utils/FromVariantConverter.java | 34 +++++++++++++++--- .../flink/table/data/utils/VariantCastUtils.java | 8 +++++ .../table/data/utils/FromVariantConverterTest.java | 40 ++++++++++++++++++++++ .../functions/casting/FromVariantCastRule.java | 6 +++- .../planner/functions/CastFunctionITCase.java | 27 ++++++++++++++- 5 files changed, 108 insertions(+), 7 deletions(-) diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java index 803728f2c22..8dd3c830652 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java @@ -79,12 +79,36 @@ public final class FromVariantConverter implements Serializable { /** * Converts a variant into internal data. * - * @throws TableRuntimeException if a value does not match its type - * @throws VariantTypeException if the variant is malformed, or a {@code BOOLEAN} value has - * another kind + * @throws TableRuntimeException if a value does not match its type, or the variant is malformed */ public Object convert(final Variant variant) { - return reader.read(variant); + try { + return reader.read(variant); + } catch (VariantTypeException e) { + // The readers check the kind first, so only an unreadable encoding gets here. + throw unreadable(e); + } + } + + /** + * Returns whether a variant is a variant null, which a cast to a nullable type reads as SQL + * {@code NULL}. A cast checks this before {@link #convert}. + * + * @throws TableRuntimeException if the variant is malformed + */ + public static boolean isVariantNull(final Variant variant) { + try { + return variant.isNull(); + } catch (VariantTypeException e) { + throw unreadable(e); + } + } + + private static TableRuntimeException unreadable(final VariantTypeException cause) { + return new TableRuntimeException( + "Cannot cast a VARIANT value whose binary encoding is malformed or uses an " + + "unknown type.", + cause); } @FunctionalInterface @@ -97,7 +121,7 @@ public final class FromVariantConverter implements Serializable { final LogicalType type, final TimeZone sessionZone, final boolean copyVariants) { switch (type.getTypeRoot()) { case BOOLEAN: - return Variant::getBoolean; + return VariantCastUtils::toBoolean; case TINYINT: return variant -> (byte) diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java index c8332182c47..26cd8cc1591 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java @@ -127,6 +127,14 @@ public final class VariantCastUtils { variant.getType(), targetType, required)); } + /** Reads a boolean variant as the target {@code BOOLEAN}. */ + public static boolean toBoolean(Variant variant) { + if (variant.getType() == Variant.Type.BOOLEAN) { + return variant.getBoolean(); + } + throw unsupportedKind(variant, "BOOLEAN"); + } + /** * Reads a numeric variant as a {@code long} and checks it against the target range. An * approximate or decimal value is accepted only when it is already integral, so nothing is diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java index 70f9807c139..2c0cd03413a 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java @@ -40,6 +40,7 @@ import org.apache.flink.table.types.logical.utils.UuidUtils; import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; +import org.apache.flink.types.variant.VariantTypeException; import org.apache.flink.util.InstantiationUtil; import org.junit.jupiter.api.Test; @@ -55,6 +56,7 @@ import java.time.LocalDateTime; import java.time.LocalTime; import java.time.ZoneId; import java.time.ZoneOffset; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -217,6 +219,44 @@ class FromVariantConverterTest { .hasMessageStartingWith("Cannot cast a VARIANT OBJECT value to VARBINARY(2)."); } + @Test + void testBooleanTargetRejectsAnotherKind() { + assertThatThrownBy(() -> convert(DataTypes.BOOLEAN(), BUILDER.of("true"))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a VARIANT STRING value to BOOLEAN."); + } + + @Test + void testMalformedVariantFails() { + final BinaryVariant valid = (BinaryVariant) BUILDER.of(1); + // The header of an INT without its four payload bytes. + final Variant truncated = + new BinaryVariant(Arrays.copyOf(valid.getValue(), 1), valid.getMetadata()); + + assertThatThrownBy(() -> convert(DataTypes.INT(), truncated)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a VARIANT value whose binary encoding is malformed or uses " + + "an unknown type.") + .hasCauseInstanceOf(VariantTypeException.class); + } + + @Test + void testNullCheckOfAnUnreadableVariantFails() { + // A primitive header with type id 31, which no version defines. + final Variant unknown = + new BinaryVariant(new byte[] {0x7C}, ((BinaryVariant) BUILDER.of(1)).getMetadata()); + + assertThatThrownBy(() -> FromVariantConverter.isVariantNull(unknown)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a VARIANT value whose binary encoding is malformed or uses " + + "an unknown type.") + .hasCauseInstanceOf(VariantTypeException.class); + assertThat(FromVariantConverter.isVariantNull(BUILDER.ofNull())).isTrue(); + assertThat(FromVariantConverter.isVariantNull(BUILDER.of(1))).isFalse(); + } + @Test void testMapKeysFollowTheLengthOfTheKeyType() { final Variant variant = BUILDER.object().add("ab", BUILDER.of(1)).build(); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java index 14985f60337..cf5b5274de8 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java @@ -77,7 +77,11 @@ class FromVariantCastRule extends AbstractNullAwareCodeGeneratorCastRule<Variant context, inputTerm, inputIsNullTerm, inputLogicalType, targetLogicalType); } final String isNullTerm = - "(" + inputIsNullTerm + " || " + methodCall(inputTerm, "isNull") + ")"; + "(" + + inputIsNullTerm + + " || " + + staticCall(FromVariantConverter.class, "isVariantNull", inputTerm) + + ")"; return super.generateCodeBlock( context, inputTerm, isNullTerm, inputLogicalType, targetLogicalType); } 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 931a0b029da..b27a3da62a9 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 @@ -32,6 +32,7 @@ import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; import org.apache.flink.types.Row; import org.apache.flink.types.bitmap.Bitmap; +import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; @@ -298,6 +299,10 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { "CAST(PARSE_JSON('true') AS BOOLEAN)", true, BOOLEAN()) + .testSqlRuntimeError( + "CAST(PARSE_JSON('\"true\"') AS BOOLEAN)", + TableRuntimeException.class, + "Cannot cast a VARIANT STRING value to BOOLEAN.") // A string reaches a binary target as its UTF-8 bytes, like a regular cast // from a character string. Any other kind has no binary form. .testResult( @@ -454,7 +459,19 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { lit("42").tryParseJson().cast(TINYINT()), "CAST(TRY_PARSE_JSON('42') AS TINYINT)", (byte) 42, - TINYINT())); + TINYINT()), + // A nullable target checks for a variant null before it casts, and that check + // fails like the cast itself. Only a source or a function can produce an + // unreadable variant. + TestSetSpec.forExpression("Cast an unreadable VARIANT to a primitive") + .onFieldsWithData(0) + .andDataTypes(INT()) + .withFunction(UnknownTypeVariantFunction.class) + .testSqlRuntimeError( + "CAST(UnknownTypeVariantFunction(f0) AS INT)", + TableRuntimeException.class, + "Cannot cast a VARIANT value whose binary encoding is malformed " + + "or uses an unknown type.")); } private static List<TestSetSpec> primitiveToVariantCasts() { @@ -2710,4 +2727,12 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { return millisOfDay; } } + + public static class UnknownTypeVariantFunction extends ScalarFunction { + public Variant eval(Integer ignored) { + // A primitive header with type id 31, which no version defines, and an empty + // dictionary. + return new BinaryVariant(new byte[] {0x7C}, new byte[] {0x01, 0x00, 0x00}); + } + } }
