This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5a1e6202832 [FLINK-40493][table] Support casting from `VARIANT` to
`UUID`
5a1e6202832 is described below
commit 5a1e6202832a4573d2605ebf31542f17c7890b81
Author: Moritz Manner <[email protected]>
AuthorDate: Tue Sep 15 08:56:44 2026 +0200
[FLINK-40493][table] Support casting from `VARIANT` to `UUID`
---
docs/content.zh/docs/sql/reference/data-types.md | 1 +
docs/content/docs/sql/reference/data-types.md | 1 +
.../types/logical/utils/LogicalTypeCasts.java | 6 ++-
.../flink/table/types/logical/utils/UuidUtils.java | 53 ++++++++++++++++++++++
.../flink/table/types/LogicalTypeCastsTest.java | 7 +--
.../casting/VariantToPrimitiveCastRule.java | 5 ++
.../table/planner/plan/utils/RexLiteralUtil.java | 8 +---
.../functions/casting/CastRuleProviderTest.java | 2 +
.../planner/functions/casting/CastRulesTest.java | 11 +++++
.../table/runtime/functions/UuidCastUtils.java | 7 +--
.../table/runtime/functions/VariantCastUtils.java | 15 ++++++
.../table/data/conversion/UuidUuidConverter.java | 12 ++---
12 files changed, 104 insertions(+), 24 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index 26fc40150b2..8ff0137dd48 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -1558,6 +1558,7 @@ to make it fit. Otherwise `CAST` fails and `TRY_CAST`
returns `NULL`.
| `TIMESTAMP` | `TIMESTAMP(p)` |
| `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)` |
| `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string |
+| `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 1a687f6b972..768c2e1ab7d 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -1566,6 +1566,7 @@ to make it fit. Otherwise `CAST` fails and `TRY_CAST`
returns `NULL`.
| `TIMESTAMP` | `TIMESTAMP(p)` |
| `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)` |
| `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string |
+| `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/types/logical/utils/LogicalTypeCasts.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
index e213c477164..d351a36746d 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
@@ -410,7 +410,11 @@ public final class LogicalTypeCasts {
// A UUID can be parsed from a character string and reinterpreted from
its 16-byte encoding.
// The reverse direction (UUID to CHAR/VARCHAR and to
BINARY(16)/BYTES) is declared on the
// respective target types.
- castTo(UUID).implicitFrom(UUID).explicitFromFamily(CHARACTER_STRING,
BINARY_STRING).build();
+ castTo(UUID)
+ .implicitFrom(UUID)
+ .explicitFromFamily(CHARACTER_STRING, BINARY_STRING)
+ .explicitFrom(VARIANT)
+ .build();
}
/**
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/UuidUtils.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/UuidUtils.java
new file mode 100644
index 00000000000..653b708760e
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/UuidUtils.java
@@ -0,0 +1,53 @@
+/*
+ * 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.types.logical.utils;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.types.logical.UuidType;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.UUID;
+
+/**
+ * Helpers to convert between a {@link UUID} and the canonical 16-byte
big-endian encoding used to
+ * store a {@link UuidType} value: the most significant 64 bits followed by
the least significant 64
+ * bits.
+ */
+@Internal
+public final class UuidUtils {
+
+ private UuidUtils() {}
+
+ /** Encodes a {@link UUID} into its canonical 16-byte big-endian
representation. */
+ public static byte[] toBytes(UUID uuid) {
+ final ByteBuffer buffer = ByteBuffer.allocate(UuidType.BYTE_LENGTH);
+ buffer.order(ByteOrder.BIG_ENDIAN);
+ buffer.putLong(uuid.getMostSignificantBits());
+ buffer.putLong(uuid.getLeastSignificantBits());
+ return buffer.array();
+ }
+
+ /** Decodes the canonical 16-byte big-endian representation into a {@link
UUID}. */
+ public static UUID fromBytes(byte[] bytes) {
+ final ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ buffer.order(ByteOrder.BIG_ENDIAN);
+ return new UUID(buffer.getLong(), buffer.getLong());
+ }
+}
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 f696227034e..1e96b65a130 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
@@ -290,6 +290,7 @@ class LogicalTypeCastsTest {
Arguments.of(new VariantType(), new CharType(), false, true),
Arguments.of(new VariantType(), VarCharType.STRING_TYPE,
false, true),
Arguments.of(new VariantType(), new TimeType(), false, true),
+ Arguments.of(new VariantType(), new UuidType(), false, true),
// variant identity cast is implicit
Arguments.of(new VariantType(), new VariantType(), true, true),
// A variant imposes a schema on a constructed target,
explicit only, recursing on
@@ -386,11 +387,11 @@ class LogicalTypeCastsTest {
Arguments.of(new BinaryType(10), new UuidType(), false, false),
// UUID identity cast is implicit
Arguments.of(new UuidType(), new UuidType(), true, true),
- // numeric and VARIANT are not castable to or from UUID
+ // numeric is not castable to or from UUID
Arguments.of(new UuidType(), new IntType(), false, false),
Arguments.of(new IntType(), new UuidType(), false, false),
- Arguments.of(new UuidType(), new VariantType(), false, false),
- Arguments.of(new VariantType(), new UuidType(), false, false));
+ // a UUID does not cast to VARIANT
+ Arguments.of(new UuidType(), new VariantType(), 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/VariantToPrimitiveCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java
index 4a16d7fce81..b0bf6703e64 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
@@ -75,6 +75,7 @@ class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<
case TIME_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case UUID:
return true;
default:
return false;
@@ -201,6 +202,10 @@ class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<
inputTerm,
LogicalTypeChecks.getPrecision(targetLogicalType)));
break;
+ case UUID:
+ writer.assignStmt(
+ returnVariable, staticCall(VariantCastUtils.class,
"toUuid", inputTerm));
+ break;
default:
throw new IllegalArgumentException(
"Unsupported target type for casting from VARIANT: " +
targetLogicalType);
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/RexLiteralUtil.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/RexLiteralUtil.java
index 2e3561f3d5f..ecabff65235 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/RexLiteralUtil.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/RexLiteralUtil.java
@@ -28,6 +28,7 @@ import
org.apache.flink.table.planner.codegen.CodeGenException;
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
import org.apache.calcite.avatica.util.ByteString;
import org.apache.calcite.rel.type.RelDataType;
@@ -39,7 +40,6 @@ import org.apache.calcite.util.TimeString;
import org.apache.calcite.util.TimestampString;
import java.math.BigDecimal;
-import java.nio.ByteBuffer;
import java.time.ZoneOffset;
import static
org.apache.flink.table.planner.utils.TimestampStringUtils.toLocalDateTime;
@@ -105,11 +105,7 @@ public class RexLiteralUtil {
break;
case UUID:
if (value instanceof java.util.UUID) {
- final java.util.UUID uuidValue = (java.util.UUID) value;
- return ByteBuffer.allocate(16)
- .putLong(uuidValue.getMostSignificantBits())
- .putLong(uuidValue.getLeastSignificantBits())
- .array();
+ return UuidUtils.toBytes((java.util.UUID) value);
}
break;
case DECIMAL:
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 0bebbd2c815..df910dfe1c4 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
@@ -47,6 +47,7 @@ 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.UUID;
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;
@@ -135,6 +136,7 @@ class CastRuleProviderTest {
assertThat(CastRuleProvider.exists(VARIANT,
TIMESTAMP_LTZ().getLogicalType())).isTrue();
assertThat(CastRuleProvider.exists(VARIANT,
TIME().getLogicalType())).isTrue();
assertThat(CastRuleProvider.exists(VARIANT,
BYTES().getLogicalType())).isTrue();
+ assertThat(CastRuleProvider.exists(VARIANT,
UUID().getLogicalType())).isTrue();
assertThat(CastRuleProvider.canFail(VARIANT, INT)).isTrue();
// INTERVAL has no VARIANT counterpart, so it is not a castable target
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 2aa04aeab2d..09c0ef5c10d 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
@@ -168,6 +168,7 @@ class CastRulesTest {
private static final Bitmap DEFAULT_BITMAP = Bitmap.fromArray(new int[]
{0, 1, 2});
private static final String UUID_STRING =
"550e8400-e29b-41d4-a716-446655440000";
+ private static final UUID UUID_VALUE = UUID.fromString(UUID_STRING);
private static final byte[] UUID_BYTES = uuidBytes(UUID_STRING);
private static byte[] uuidBytes(String uuid) {
@@ -1864,6 +1865,12 @@ class CastRulesTest {
Variant.newBuilder()
.of(Instant.parse("2021-09-24T12:34:56.123456789Z")),
fromString("2021-09-24 14:34:56.123456789"))
+ // a UUID renders in its canonical lower-case
8-4-4-4-12 form, the same as
+ // a regular UUID to string cast
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(UUID_VALUE),
+ fromString(UUID_STRING))
// a binary value is read as UTF-8, like a regular
BINARY to string cast
.fromCase(
VARIANT(),
@@ -2034,6 +2041,10 @@ class CastRulesTest {
// 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),
+ CastTestSpecBuilder.testCastTo(UUID())
+ .fromCase(VARIANT(), null, null)
+ .fromCase(VARIANT(),
Variant.newBuilder().of(UUID_VALUE), UUID_BYTES)
+ .fail(VARIANT(), VARIANT_BUILDER.of("foo"),
TableRuntimeException.class),
CastTestSpecBuilder.testCastTo(DATE())
.fromCase(
VARIANT(),
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/UuidCastUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/UuidCastUtils.java
index 79a4921caca..3cd5ce75dee 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/UuidCastUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/UuidCastUtils.java
@@ -21,9 +21,7 @@ package org.apache.flink.table.runtime.functions;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.types.logical.UuidType;
-
-import java.nio.ByteBuffer;
-import java.util.UUID;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
/**
* Runtime helpers for casting to and from a {@code UUID} value.
@@ -42,8 +40,7 @@ public final class UuidCastUtils {
* the 16-byte big-endian encoding, so this never fails.
*/
public static String toStringValue(byte[] uuidBytes) {
- final ByteBuffer buffer = ByteBuffer.wrap(uuidBytes);
- return new UUID(buffer.getLong(), buffer.getLong()).toString();
+ return UuidUtils.fromBytes(uuidBytes).toString();
}
/**
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
index 561da100cee..0405575aaed 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.data.binary.BinaryStringDataUtil;
import org.apache.flink.table.data.binary.StringUtf8Utils;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.variant.Variant;
@@ -271,6 +272,17 @@ public final class VariantCastUtils {
throw unsupportedKind(variant, String.format("TIME(%d)", precision));
}
+ /**
+ * Reads a UUID variant as the target {@code UUID}, encoded as the
canonical 16-byte big-endian
+ * representation.
+ */
+ public static byte[] toUuid(Variant variant) {
+ if (variant.getType() == Variant.Type.UUID) {
+ return UuidUtils.toBytes(variant.getUuid());
+ }
+ throw unsupportedKind(variant, "UUID");
+ }
+
/**
* 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
@@ -433,6 +445,9 @@ public final class VariantCastUtils {
sessionZone,
TIMESTAMP_NANOS_PRECISION);
break;
+ case UUID:
+ value = variant.getUuid().toString();
+ break;
case NULL:
// Only reachable for a NOT NULL target. A nullable target
maps a null-valued
// variant to SQL NULL before this method is called.
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/UuidUuidConverter.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/UuidUuidConverter.java
index a3f97f9da6c..efb04d0228a 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/UuidUuidConverter.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/UuidUuidConverter.java
@@ -20,8 +20,8 @@ package org.apache.flink.table.data.conversion;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.types.logical.UuidType;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
-import java.nio.ByteBuffer;
import java.util.UUID;
/**
@@ -37,17 +37,11 @@ public class UuidUuidConverter implements
DataStructureConverter<byte[], UUID> {
@Override
public byte[] toInternal(UUID external) {
- return ByteBuffer.allocate(16)
- .putLong(external.getMostSignificantBits())
- .putLong(external.getLeastSignificantBits())
- .array();
+ return UuidUtils.toBytes(external);
}
@Override
public UUID toExternal(byte[] internal) {
- final ByteBuffer buffer = ByteBuffer.wrap(internal);
- final long mostSignificantBits = buffer.getLong();
- final long leastSignificantBits = buffer.getLong();
- return new UUID(mostSignificantBits, leastSignificantBits);
+ return UuidUtils.fromBytes(internal);
}
}
