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 7c5b2379cb5d117ed26648f14bae4c9f11649e77 Author: Ramin Gharib <[email protected]> AuthorDate: Fri Nov 6 10:19:19 2026 +0100 [FLINK-40319][table] Allow CAST from a VARIANT string value to BINARY and VARBINARY A cast from VARIANT to BINARY, VARBINARY or BYTES read the value with Variant#getBytes, which accepts only the BYTES kind. A string variant failed whatever it held, and so did every other kind, with the getter's VariantTypeException instead of the TableRuntimeException that the other readers throw. A string now reaches a binary target as its UTF-8 bytes, padded or truncated like a stored BYTES value. This matches a regular cast from a character string and mirrors the cast from a BYTES variant to a character string. JSON has no binary type, so it also lets a variant built by PARSE_JSON reach a binary target at all. Every other kind fails with the unsupportedKind message, as regular SQL also rejects BOOLEAN or INTEGER to BINARY. The bytes come from getString, so a stored string whose payload is not valid UTF-8 comes out as the bytes of U+FFFD, the same substitution the cast to a character string makes. No raw-bytes accessor is added. CastRulesTest asserted that a string variant fails a BYTES cast. That case now passes. Generated-by: Claude Code 2.1.319 (Claude Opus 5.5) --- docs/content.zh/docs/sql/reference/data-types.md | 1 + docs/content/docs/sql/reference/data-types.md | 1 + .../table/data/utils/FromVariantConverter.java | 4 +- .../flink/table/data/utils/VariantCastUtils.java | 31 ++++++++++++++-- .../table/data/utils/FromVariantConverterTest.java | 43 ++++++++++++++++++++++ .../planner/functions/CastFunctionITCase.java | 16 ++++++++ .../planner/functions/casting/CastRulesTest.java | 3 +- 7 files changed, 92 insertions(+), 7 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index cb310a235fc..e36eb8d9ea3 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1565,6 +1565,7 @@ and cast it by the same rules, returning `NULL` when nothing is at the path. | `TIMESTAMP` | `TIMESTAMP(p)`, `TIMESTAMP_LTZ(p)`, `DATE`, `TIME(p)` | | `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)`, `TIMESTAMP(p)`, `DATE`, `TIME(p)` | | `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string | +| `STRING` | `BINARY(n)`, `VARBINARY(n)` as its UTF-8 bytes | | `UUID` | `UUID` | | any scalar | `STRING`, `CHAR(n)`, `VARCHAR(n)` | | `NULL` | SQL `NULL` for any nullable target | diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index b0d6b95fd82..d5e3d765d40 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1573,6 +1573,7 @@ and cast it by the same rules, returning `NULL` when nothing is at the path. | `TIMESTAMP` | `TIMESTAMP(p)`, `TIMESTAMP_LTZ(p)`, `DATE`, `TIME(p)` | | `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)`, `TIMESTAMP(p)`, `DATE`, `TIME(p)` | | `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string | +| `STRING` | `BINARY(n)`, `VARBINARY(n)` as its UTF-8 bytes | | `UUID` | `UUID` | | any scalar | `STRING`, `CHAR(n)`, `VARCHAR(n)` | | `NULL` | SQL `NULL` for any nullable target | 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 8056eac0df6..803728f2c22 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 @@ -80,8 +80,8 @@ 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} or binary - * value has another kind + * @throws VariantTypeException if the variant is malformed, or a {@code BOOLEAN} value has + * another kind */ public Object convert(final Variant variant) { return reader.read(variant); 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 532505ec3da..c8332182c47 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 @@ -351,12 +351,25 @@ public final class VariantCastUtils { } /** - * Reads a binary variant as the target binary type. {@code BINARY} is fixed width, so a shorter - * value is padded with zero bytes, and either target truncates a value longer than {@code - * targetLength}. This matches a regular cast into the same type. + * Reads a binary or string variant as the target binary type. A string becomes its UTF-8 bytes, + * like a regular cast from a character string. The string is decoded first, so a payload that + * is not valid UTF-8 comes out with the bytes of U+FFFD in place of each malformed sequence, as + * in the cast to a character string. {@code BINARY} is fixed width, so a shorter value is + * padded with zero bytes, and either target truncates a value longer than {@code targetLength}. + * This matches a regular cast into the same type. */ public static byte[] toBytes(Variant variant, int targetLength, boolean fixedLength) { - final byte[] value = variant.getBytes(); + final byte[] value; + switch (variant.getType()) { + case BYTES: + value = variant.getBytes(); + break; + case STRING: + value = variant.getString().getBytes(StandardCharsets.UTF_8); + break; + default: + throw unsupportedKind(variant, binaryTarget(targetLength, fixedLength)); + } if (fixedLength) { return value.length == targetLength ? value : Arrays.copyOf(value, targetLength); } @@ -587,6 +600,16 @@ public final class VariantCastUtils { return String.format("%s(%d)", "VARCHAR", targetLength); } + private static String binaryTarget(int targetLength, boolean fixedLength) { + if (fixedLength) { + return String.format("%s(%d)", "BINARY", targetLength); + } + if (targetLength == Integer.MAX_VALUE) { + return "BYTES"; + } + return String.format("%s(%d)", "VARBINARY", targetLength); + } + private static Number numeric(Variant variant, String targetType) { switch (variant.getType()) { case TINYINT: 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 d98aebe1cf0..70f9807c139 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 @@ -48,6 +48,7 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; @@ -174,6 +175,48 @@ class FromVariantConverterTest { .hasMessage("Casting the VARIANT value 9223372036854775808 to BIGINT overflowed."); } + @Test + void testStringReachesABinaryTargetAsItsUtf8Bytes() { + final Variant variant = BUILDER.of("héllo"); + + assertThat((byte[]) convert(DataTypes.BYTES(), variant)) + .isEqualTo("héllo".getBytes(StandardCharsets.UTF_8)); + assertThat((byte[]) convert(DataTypes.VARBINARY(3), variant)) + .containsExactly(0x68, 0xC3, 0xA9); + assertThat((byte[]) convert(DataTypes.BINARY(8), variant)) + .containsExactly(0x68, 0xC3, 0xA9, 0x6C, 0x6C, 0x6F, 0, 0); + } + + @Test + void testStringWithInvalidUtf8ReachesABinaryTargetWithReplacementCharacters() { + final BinaryVariant valid = (BinaryVariant) BUILDER.of("ab"); + final byte[] value = valid.getValue().clone(); + // The two payload bytes after the header become C3 28, which is not valid UTF-8. + value[1] = (byte) 0xC3; + value[2] = 0x28; + final Variant invalid = new BinaryVariant(value, valid.getMetadata()); + + assertThat((byte[]) convert(DataTypes.BYTES(), invalid)) + .containsExactly(0xEF, 0xBF, 0xBD, 0x28); + } + + @Test + void testBinaryTargetRejectsAKindWithoutBytes() { + assertThatThrownBy(() -> convert(DataTypes.BYTES(), BUILDER.of(true))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a VARIANT BOOLEAN value to BYTES."); + assertThatThrownBy(() -> convert(DataTypes.BINARY(2), BUILDER.of(1))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a VARIANT INT value to BINARY(2)."); + assertThatThrownBy( + () -> + convert( + DataTypes.VARBINARY(2), + BUILDER.object().add("a", BUILDER.of(1)).build())) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a VARIANT OBJECT value to VARBINARY(2)."); + } + @Test void testMapKeysFollowTheLengthOfTheKeyType() { final Variant variant = BUILDER.object().add("ab", BUILDER.of(1)).build(); 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 abd3d48ef81..931a0b029da 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 @@ -298,6 +298,22 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { "CAST(PARSE_JSON('true') AS BOOLEAN)", true, 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( + lit("\"abc\"").parseJson().cast(BYTES()), + "CAST(PARSE_JSON('\"abc\"') AS BYTES)", + new byte[] {'a', 'b', 'c'}, + BYTES()) + .testSqlRuntimeError( + "CAST(PARSE_JSON('true') AS BYTES)", + TableRuntimeException.class, + "Cannot cast a VARIANT BOOLEAN value to BYTES.") + .testResult( + lit("true").parseJson().tryCast(BYTES()), + "TRY_CAST(PARSE_JSON('true') AS BYTES)", + null, + BYTES()) // CAST returns the raw scalar value (string unquoted) .testResult( lit("\"foo\"").parseJson().cast(STRING()), 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 1b05d848899..9410ba66f3b 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 @@ -2306,7 +2306,8 @@ class CastRulesTest { // the raw bytes stay reachable when the character string cast rejects // them, which is what makes this the way to inspect such a value .fromCase(VARIANT(), VARIANT_BUILDER.of(INVALID_UTF8), INVALID_UTF8) - .fail(VARIANT(), VARIANT_BUILDER.of("foo"), TableRuntimeException.class), + .fromCase(VARIANT(), VARIANT_BUILDER.of("ab"), new byte[] {'a', 'b'}) + .fail(VARIANT(), VARIANT_BUILDER.of(true), TableRuntimeException.class), CastTestSpecBuilder.testCastTo(UUID()) .fromCase(VARIANT(), null, null) .fromCase(VARIANT(), Variant.newBuilder().of(UUID_VALUE), UUID_BYTES)
