This is an automated email from the ASF dual-hosted git repository. raminqaf pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit de2cb88cb0c5a1c076364e02dd03a19b6d890b91 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 and VARIANT_GET first check for a variant null. That check reads the type outside convert and threw the same VariantTypeException. Both now call FromVariantConverter#isVariantNull, which wraps it the same way, so TRY_VARIANT_GET returns NULL for such a value instead of throwing. Generated-by: Claude Code 2.1.319 (Claude Opus 5.5) --- .../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 ++++++++++++++- .../planner/functions/VariantFunctionsITCase.java | 16 ++++++++- .../functions/scalar/VariantGetFunction.java | 6 ++-- 7 files changed, 126 insertions(+), 11 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}); + } + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java index 6780b01b5cc..4691f8aeeba 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/VariantFunctionsITCase.java @@ -56,7 +56,7 @@ class VariantFunctionsITCase extends BuiltInFunctionTestBase { @Override Stream<TestSetSpec> getTestSetSpecs() { - return Stream.of(variantGet(), variantGetFailure(), tryVariantGet()); + return Stream.of(variantGet(), variantGetFailure(), tryVariantGet(), unreadableVariant()); } private static TestSetSpec variantGet() { @@ -164,4 +164,18 @@ class VariantFunctionsITCase extends BuiltInFunctionTestBase { "Invalid path '$.user[': expected an index or a quoted name after " + "'$.user['."); } + + private static TestSetSpec unreadableVariant() { + return TestSetSpec.forFunction(BuiltInFunctionDefinitions.TRY_VARIANT_GET, "unreadable") + .onFieldsWithData(0) + .andDataTypes(INT()) + .withFunction(CastFunctionITCase.UnknownTypeVariantFunction.class) + .testSqlResult( + "TRY_VARIANT_GET(UnknownTypeVariantFunction(f0), '$', 'INT')", null, INT()) + .testSqlRuntimeError( + "VARIANT_GET(UnknownTypeVariantFunction(f0), '$', 'INT')", + TableRuntimeException.class, + "Cannot cast a VARIANT value whose binary encoding is malformed or uses " + + "an unknown type."); + } } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java index d6d1facbb45..e25699e8c21 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/VariantGetFunction.java @@ -104,10 +104,10 @@ public class VariantGetFunction extends BuiltInScalarFunction { if (value == null || converter == null) { return value; } - if (value.isNull()) { - return null; - } try { + if (FromVariantConverter.isVariantNull(value)) { + return null; + } return converter.convert(value); } catch (RuntimeException e) { if (isTry) {
