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 995865e08a7 [FLINK-40490][table] Add runtime serialization and codegen
for the `UUID` type
995865e08a7 is described below
commit 995865e08a7633df7a403abedccf8f27747cdf84
Author: Ramin Gharib <[email protected]>
AuthorDate: Fri Sep 4 09:09:33 2026 +0200
[FLINK-40490][table] Add runtime serialization and codegen for the `UUID`
type
---
.../flink/api/common/typeinfo/BasicTypeInfo.java | 10 ++
.../apache/flink/api/common/typeinfo/Types.java | 4 +
.../api/common/typeutils/base/UuidComparator.java | 144 +++++++++++++++++++++
.../api/common/typeutils/base/UuidSerializer.java | 112 ++++++++++++++++
.../common/typeutils/base/UuidComparatorTest.java | 59 +++++++++
.../common/typeutils/base/UuidSerializerTest.java | 55 ++++++++
.../org/apache/flink/table/data/ArrayData.java | 1 +
.../java/org/apache/flink/table/data/RowData.java | 3 +
.../flink/table/data/binary/BinaryArrayData.java | 1 +
.../apache/flink/table/types/logical/UuidType.java | 3 +-
.../types/logical/utils/LogicalTypeUtils.java | 1 +
.../table/types/utils/ClassDataTypeConverter.java | 1 +
.../utils/LegacyTypeInfoDataTypeConverter.java | 3 +
.../types/utils/TypeInfoDataTypeConverter.java | 2 +
.../table/types/ClassDataTypeConverterTest.java | 1 +
.../apache/flink/table/types/LogicalTypesTest.java | 4 +-
.../types/extraction/DataTypeExtractorTest.java | 3 +
.../table/planner/plan/utils/RexLiteralUtil.java | 10 ++
.../flink/table/planner/codegen/CodeGenUtils.scala | 8 +-
.../table/planner/codegen/ExpressionReducer.scala | 6 +-
.../table/planner/codegen/GenerateUtils.scala | 2 +-
.../planner/plan/metadata/FlinkRelMdSize.scala | 2 +-
.../plan/nodes/exec/stream/UuidSemanticTest.java | 40 ++++++
.../plan/nodes/exec/stream/UuidTestPrograms.java | 115 ++++++++++++++++
.../data/conversion/DataStructureConverters.java | 3 +
.../table/data/conversion/UuidUuidConverter.java | 53 ++++++++
.../flink/table/data/writer/BinaryArrayWriter.java | 1 +
.../flink/table/data/writer/BinaryWriter.java | 2 +
.../runtime/typeutils/InternalSerializers.java | 1 +
.../table/data/DataStructureConvertersTest.java | 5 +
.../TypeSerializerTestCoverageTest.java | 4 +-
31 files changed, 646 insertions(+), 13 deletions(-)
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
index afe11eb110f..a3bbdf9889b 100644
---
a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
+++
b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
@@ -51,6 +51,8 @@ import
org.apache.flink.api.common.typeutils.base.ShortComparator;
import org.apache.flink.api.common.typeutils.base.ShortSerializer;
import org.apache.flink.api.common.typeutils.base.StringComparator;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.common.typeutils.base.UuidComparator;
+import org.apache.flink.api.common.typeutils.base.UuidSerializer;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import java.lang.reflect.Constructor;
@@ -62,6 +64,7 @@ import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -160,6 +163,9 @@ public class BasicTypeInfo<T> extends TypeInformation<T>
implements AtomicType<T
new Class<?>[] {},
InstantSerializer.INSTANCE,
InstantComparator.class);
+ public static final BasicTypeInfo<UUID> UUID_TYPE_INFO =
+ new BasicTypeInfo<>(
+ UUID.class, new Class<?>[] {}, UuidSerializer.INSTANCE,
UuidComparator.class);
//
--------------------------------------------------------------------------------------------
@@ -338,5 +344,9 @@ public class BasicTypeInfo<T> extends TypeInformation<T>
implements AtomicType<T
TYPES.put(BigInteger.class, BIG_INT_TYPE_INFO);
TYPES.put(BigDecimal.class, BIG_DEC_TYPE_INFO);
TYPES.put(Instant.class, INSTANT_TYPE_INFO);
+ // UUID is intentionally not registered here. Automatic reflective
extraction of
+ // java.util.UUID to UUID_TYPE_INFO in the DataStream API stays opt-in
via Types.UUID to
+ // avoid silently changing the serializer of existing java.util.UUID
fields (currently
+ // handled by Kryo). This is planned to change in the next major Flink
version.
}
}
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
index 52b4f827b53..32b6925d12b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
@@ -56,6 +56,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
/**
* This class gives access to the type information of the most common types
for which Flink has
@@ -158,6 +159,9 @@ public class Types {
/** Returns type information for {@link java.time.Instant}. Supports a
null value. */
public static final TypeInformation<Instant> INSTANT =
BasicTypeInfo.INSTANT_TYPE_INFO;
+ /** Returns type information for {@link java.util.UUID}. */
+ public static final TypeInformation<UUID> UUID =
BasicTypeInfo.UUID_TYPE_INFO;
+
public static final TypeInformation<Variant> VARIANT =
VariantTypeInfo.INSTANCE;
/**
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidComparator.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidComparator.java
new file mode 100644
index 00000000000..f190aa03b4c
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidComparator.java
@@ -0,0 +1,144 @@
+/*
+ * 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.api.common.typeutils.base;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.MemorySegment;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * Comparator for {@link UUID}.
+ *
+ * <p>Ordering follows the canonical unsigned big-endian byte comparison of
the 16-byte encoding, so
+ * every 128-bit value sorts by its raw bytes. This deliberately differs from
{@link
+ * UUID#compareTo(UUID)}, which compares the two 64-bit halves as signed
longs; the unsigned order
+ * is the one defined by the type and used consistently across serialized,
normalized-key, and
+ * object comparisons.
+ */
+@Internal
+public final class UuidComparator extends BasicTypeComparator<UUID> {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient UUID reference;
+
+ public UuidComparator(boolean ascending) {
+ super(ascending);
+ }
+
+ private static int compareUnsigned(UUID first, UUID second) {
+ int comp =
+ Long.compareUnsigned(
+ first.getMostSignificantBits(),
second.getMostSignificantBits());
+ if (comp == 0) {
+ comp =
+ Long.compareUnsigned(
+ first.getLeastSignificantBits(),
second.getLeastSignificantBits());
+ }
+ return comp;
+ }
+
+ @Override
+ public void setReference(UUID toCompare) {
+ super.setReference(toCompare);
+ this.reference = toCompare;
+ }
+
+ @Override
+ public int compareToReference(TypeComparator<UUID> referencedComparator) {
+ // Mirror BasicTypeComparator's inverted operand order (referenced vs.
this) using the
+ // unsigned ordering, so it stays consistent with
compare/compareSerialized.
+ final int comp =
+ compareUnsigned(((UuidComparator)
referencedComparator).reference, this.reference);
+ return ascendingComparison ? comp : -comp;
+ }
+
+ @Override
+ public int compare(UUID first, UUID second) {
+ final int comp = compareUnsigned(first, second);
+ return ascendingComparison ? comp : -comp;
+ }
+
+ @Override
+ public int compareSerialized(DataInputView firstSource, DataInputView
secondSource)
+ throws IOException {
+ final long lMostSignificantBits = firstSource.readLong();
+ final long rMostSignificantBits = secondSource.readLong();
+ int comp = Long.compareUnsigned(lMostSignificantBits,
rMostSignificantBits);
+ if (comp == 0) {
+ final long lLeastSignificantBits = firstSource.readLong();
+ final long rLeastSignificantBits = secondSource.readLong();
+ comp = Long.compareUnsigned(lLeastSignificantBits,
rLeastSignificantBits);
+ }
+ return ascendingComparison ? comp : -comp;
+ }
+
+ @Override
+ public boolean supportsNormalizedKey() {
+ return true;
+ }
+
+ @Override
+ public int getNormalizeKeyLen() {
+ return UuidSerializer.UUID_BYTES;
+ }
+
+ @Override
+ public boolean isNormalizedKeyPrefixOnly(int keyBytes) {
+ return keyBytes < getNormalizeKeyLen();
+ }
+
+ @Override
+ public void putNormalizedKey(UUID record, MemorySegment target, int
offset, int numBytes) {
+ // The raw 16-byte big-endian encoding is already order-preserving
under the unsigned
+ // byte-wise comparison used for normalized keys, so no sign flip is
applied.
+ final long mostSignificantBits = record.getMostSignificantBits();
+ final long leastSignificantBits = record.getLeastSignificantBits();
+ if (numBytes >= Long.BYTES) {
+ target.putLongBigEndian(offset, mostSignificantBits);
+ offset += Long.BYTES;
+ numBytes -= Long.BYTES;
+ if (numBytes >= Long.BYTES) {
+ target.putLongBigEndian(offset, leastSignificantBits);
+ offset += Long.BYTES;
+ numBytes -= Long.BYTES;
+ for (int i = 0; i < numBytes; i++) {
+ target.put(offset + i, (byte) 0);
+ }
+ } else {
+ for (int i = 0; i < numBytes; i++) {
+ target.put(offset + i, (byte) (leastSignificantBits >>>
((7 - i) << 3)));
+ }
+ }
+ } else {
+ for (int i = 0; i < numBytes; i++) {
+ target.put(offset + i, (byte) (mostSignificantBits >>> ((7 -
i) << 3)));
+ }
+ }
+ }
+
+ @Override
+ public TypeComparator<UUID> duplicate() {
+ return new UuidComparator(ascendingComparison);
+ }
+}
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidSerializer.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidSerializer.java
new file mode 100644
index 00000000000..e0d76c212d4
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/UuidSerializer.java
@@ -0,0 +1,112 @@
+/*
+ * 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.api.common.typeutils.base;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * Serializer for {@link UUID}.
+ *
+ * <p>A UUID is written as its canonical 16-byte big-endian form: the most
significant 64 bits
+ * followed by the least significant 64 bits. All 128 bits are valid, so there
is no reserved value
+ * for {@code null}; nullability is handled by wrapping serializers (e.g.
{@code NullableSerializer}
+ * or the row/POJO serializers) as for the other fixed-width base serializers.
+ */
+@Internal
+public final class UuidSerializer extends TypeSerializerSingleton<UUID> {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Length of a serialized UUID: two 64-bit longs. */
+ static final int UUID_BYTES = 2 * Long.BYTES;
+
+ public static final UuidSerializer INSTANCE = new UuidSerializer();
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public UUID createInstance() {
+ return new UUID(0L, 0L);
+ }
+
+ @Override
+ public UUID copy(UUID from) {
+ return from;
+ }
+
+ @Override
+ public UUID copy(UUID from, UUID reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return UUID_BYTES;
+ }
+
+ @Override
+ public void serialize(UUID record, DataOutputView target) throws
IOException {
+ target.writeLong(record.getMostSignificantBits());
+ target.writeLong(record.getLeastSignificantBits());
+ }
+
+ @Override
+ public UUID deserialize(DataInputView source) throws IOException {
+ final long mostSignificantBits = source.readLong();
+ final long leastSignificantBits = source.readLong();
+ return new UUID(mostSignificantBits, leastSignificantBits);
+ }
+
+ @Override
+ public UUID deserialize(UUID reuse, DataInputView source) throws
IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ target.writeLong(source.readLong());
+ target.writeLong(source.readLong());
+ }
+
+ @Override
+ public TypeSerializerSnapshot<UUID> snapshotConfiguration() {
+ return new UuidSerializerSnapshot();
+ }
+
+ // ------------------------------------------------------------------------
+
+ /** Serializer configuration snapshot for compatibility and format
evolution. */
+ @Internal
+ public static final class UuidSerializerSnapshot extends
SimpleTypeSerializerSnapshot<UUID> {
+
+ public UuidSerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git
a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidComparatorTest.java
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidComparatorTest.java
new file mode 100644
index 00000000000..34b67048bf9
--- /dev/null
+++
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidComparatorTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.api.common.typeutils.base;
+
+import org.apache.flink.api.common.typeutils.ComparatorTestBase;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.UUID;
+
+/**
+ * A test for the {@link UuidComparator}.
+ *
+ * <p>The sorted data spans the signed-long boundary (a {@code 0x8000...}
most-significant half
+ * sorts after {@code 0x7FFF...}) to assert the unsigned big-endian ordering,
which differs from
+ * {@link UUID#compareTo(UUID)}.
+ */
+class UuidComparatorTest extends ComparatorTestBase<UUID> {
+
+ @Override
+ protected TypeComparator<UUID> createComparator(boolean ascending) {
+ return new UuidComparator(ascending);
+ }
+
+ @Override
+ protected TypeSerializer<UUID> createSerializer() {
+ return new UuidSerializer();
+ }
+
+ @Override
+ protected UUID[] getSortedTestData() {
+ return new UUID[] {
+ new UUID(0x0000000000000000L, 0x0000000000000000L),
+ new UUID(0x0000000000000000L, 0x0000000000000001L),
+ new UUID(0x0000000000000000L, 0xFFFFFFFFFFFFFFFFL),
+ new UUID(0x0000000000000001L, 0x0000000000000000L),
+ new UUID(0x7FFFFFFFFFFFFFFFL, 0xFFFFFFFFFFFFFFFFL),
+ new UUID(0x8000000000000000L, 0x0000000000000000L),
+ new UUID(0xFFFFFFFFFFFFFFFFL, 0x0000000000000000L),
+ new UUID(0xFFFFFFFFFFFFFFFFL, 0xFFFFFFFFFFFFFFFFL)
+ };
+ }
+}
diff --git
a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidSerializerTest.java
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidSerializerTest.java
new file mode 100644
index 00000000000..9d62538a7f8
--- /dev/null
+++
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/UuidSerializerTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.api.common.typeutils.base;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.UUID;
+
+/** A test for the {@link UuidSerializer}. */
+class UuidSerializerTest extends SerializerTestBase<UUID> {
+
+ @Override
+ protected TypeSerializer<UUID> createSerializer() {
+ return new UuidSerializer();
+ }
+
+ @Override
+ protected int getLength() {
+ return 16;
+ }
+
+ @Override
+ protected Class<UUID> getTypeClass() {
+ return UUID.class;
+ }
+
+ @Override
+ protected UUID[] getTestData() {
+ return new UUID[] {
+ new UUID(0L, 0L),
+ new UUID(0x8000000000000000L, 0x0000000000000000L),
+ new UUID(0xFFFFFFFFFFFFFFFFL, 0xFFFFFFFFFFFFFFFFL),
+ UUID.fromString("00000000-0000-0000-0000-000000000001"),
+ UUID.fromString("550e8400-e29b-41d4-a716-446655440000"),
+ UUID.fromString("f47ac10b-58cc-4372-a567-0e02b2c3d479")
+ };
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
index 811c9be70e6..c56955c4a5c 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
@@ -164,6 +164,7 @@ public interface ArrayData {
break;
case BINARY:
case VARBINARY:
+ case UUID:
elementGetter = ArrayData::getBinary;
break;
case DECIMAL:
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
index ca43f1608f9..1a28ebfd833 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
@@ -111,6 +111,8 @@ import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getSc
* +--------------------------------+-----------------------------------------+
* | BITMAP | {@link Bitmap} |
* +--------------------------------+-----------------------------------------+
+ * | UUID | byte[] (16 big-endian bytes) |
+ * +--------------------------------+-----------------------------------------+
* </pre>
*
* <p>Nullability is always handled by the container data structure.
@@ -238,6 +240,7 @@ public interface RowData {
break;
case BINARY:
case VARBINARY:
+ case UUID:
fieldGetter = row -> row.getBinary(fieldPos);
break;
case DECIMAL:
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
index 10a8b3e6ef7..a8e3087d72c 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
@@ -95,6 +95,7 @@ public final class BinaryArrayData extends BinarySection
implements ArrayData, T
case RAW:
case VARIANT:
case BITMAP:
+ case UUID:
// long and double are 8 bytes;
// otherwise it stores the length and offset of the
variable-length part for types
// such as is string, map, etc.
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/UuidType.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/UuidType.java
index 2973a89b601..f18076685d6 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/UuidType.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/UuidType.java
@@ -38,7 +38,8 @@ public final class UuidType extends LogicalType {
private static final long serialVersionUID = 1L;
- private static final Set<String> INPUT_OUTPUT_CONVERSION =
conversionSet(UUID.class.getName());
+ private static final Set<String> INPUT_OUTPUT_CONVERSION =
+ conversionSet(UUID.class.getName(), byte[].class.getName());
public UuidType(boolean isNullable) {
super(isNullable, LogicalTypeRoot.UUID);
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
index 98629b15fc1..849c2f22bbf 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
@@ -71,6 +71,7 @@ public final class LogicalTypeUtils {
return Boolean.class;
case BINARY:
case VARBINARY:
+ case UUID:
return byte[].class;
case DECIMAL:
return DecimalData.class;
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
index f0fca13ea29..726843ce284 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java
@@ -79,6 +79,7 @@ public final class ClassDataTypeConverter {
addDefaultDataType(
java.time.Period.class, DataTypes.INTERVAL(DataTypes.YEAR(4),
DataTypes.MONTH()));
addDefaultDataType(ColumnList.class, DataTypes.DESCRIPTOR());
+ addDefaultDataType(java.util.UUID.class, DataTypes.UUID());
addDefaultDataType(Variant.class, DataTypes.VARIANT());
addDefaultDataType(Bitmap.class, DataTypes.BITMAP());
addDefaultDataType(RoaringBitmapData.class, DataTypes.BITMAP());
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java
index 1071a661ef1..dc82a07ba81 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java
@@ -73,6 +73,7 @@ import static
org.apache.flink.table.types.logical.LogicalTypeRoot.STRUCTURED_TY
import static
org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE;
import static
org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
import static
org.apache.flink.table.types.logical.LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.UUID;
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.isRowtimeAttribute;
@@ -228,6 +229,8 @@ public final class LegacyTypeInfoDataTypeConverter {
return Types.STRING;
} else if (logicalType.is(VARIANT)) {
return Types.VARIANT;
+ } else if (logicalType.is(UUID)) {
+ return Types.UUID;
}
// relax the precision constraint as Timestamp can store the highest
precision
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/TypeInfoDataTypeConverter.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/TypeInfoDataTypeConverter.java
index 7d8163aba53..941a8c854ef 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/TypeInfoDataTypeConverter.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/TypeInfoDataTypeConverter.java
@@ -58,6 +58,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -149,6 +150,7 @@ public final class TypeInfoDataTypeConverter {
PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO,
DataTypes.ARRAY(DataTypes.DOUBLE().notNull().bridgedTo(double.class))
.bridgedTo(double[].class));
+ conversionMap.put(Types.UUID, DataTypes.UUID().bridgedTo(UUID.class));
conversionMap.put(Types.VARIANT,
DataTypes.VARIANT().bridgedTo(Variant.class));
conversionMap.put(Types.BITMAP,
DataTypes.BITMAP().bridgedTo(Bitmap.class));
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
index ebf629a0f72..00f07e9375a 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java
@@ -94,6 +94,7 @@ class ClassDataTypeConverterTest {
TimeIntervalUnit.class,
new AtomicDataType(new
SymbolType<>()).bridgedTo(TimeIntervalUnit.class)),
of(Row.class, null),
+ of(java.util.UUID.class, DataTypes.UUID()),
of(Variant.class, DataTypes.VARIANT()),
of(Bitmap.class, DataTypes.BITMAP().bridgedTo(Bitmap.class)),
of(RoaringBitmapData.class,
DataTypes.BITMAP().bridgedTo(RoaringBitmapData.class)));
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
index 0703e16c78e..03dacc344ee 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
@@ -636,8 +636,8 @@ public class LogicalTypesTest {
baseAssertions(
"UUID",
"UUID",
- new Class[] {UUID.class},
- new Class[] {UUID.class},
+ new Class[] {UUID.class, byte[].class},
+ new Class[] {UUID.class, byte[].class},
new LogicalType[] {},
new UuidType(false)));
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
index da019284223..f5e4423b9de 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
@@ -81,6 +81,9 @@ class DataTypeExtractorTest {
// simple extraction of BYTES
TestSpec.forType(byte[].class).expectDataType(DataTypes.BYTES()),
+ // automatic extraction of UUID for Table/SQL
+
TestSpec.forType(java.util.UUID.class).expectDataType(DataTypes.UUID()),
+
// extraction from hint conversion class
TestSpec.forType(
new DataTypeHintMock() {
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 dc3875ce4a7..2e3561f3d5f 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
@@ -39,6 +39,7 @@ 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;
@@ -102,6 +103,15 @@ public class RexLiteralUtil {
return ((ByteString) value).getBytes();
}
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();
+ }
+ break;
case DECIMAL:
if (value instanceof BigDecimal) {
return DecimalData.fromBigDecimal(
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
index 739858689bf..0dd59e18235 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
@@ -263,7 +263,7 @@ object CodeGenUtils {
// ordered by type root definition
case CHAR | VARCHAR => BINARY_STRING
case BOOLEAN => className[JBoolean]
- case BINARY | VARBINARY => "byte[]"
+ case BINARY | VARBINARY | UUID => "byte[]"
case DECIMAL => className[DecimalData]
case TINYINT => className[JByte]
case SMALLINT => className[JShort]
@@ -329,7 +329,7 @@ object CodeGenUtils {
s"$term.hashCode()"
case BOOLEAN =>
s"${className[JBoolean]}.hashCode($term)"
- case BINARY | VARBINARY =>
+ case BINARY | VARBINARY | UUID =>
// Instead of computing the BYTE_ARRAY_BASE_OFFSET value in JM,
generate the code
// and evaluate it in TM. This is required so that byte array offset
will be consistent.
// See FLINK-37833 for more details.
@@ -510,7 +510,7 @@ object CodeGenUtils {
s"(($BINARY_STRING) $rowTerm.getString($indexTerm))"
case BOOLEAN =>
s"$rowTerm.getBoolean($indexTerm)"
- case BINARY | VARBINARY =>
+ case BINARY | VARBINARY | UUID =>
s"$rowTerm.getBinary($indexTerm)"
case DECIMAL =>
s"$rowTerm.getDecimal($indexTerm, ${getPrecision(t)}, ${getScale(t)})"
@@ -798,7 +798,7 @@ object CodeGenUtils {
s"$writerTerm.writeString($indexTerm, $fieldValTerm)"
case BOOLEAN =>
s"$writerTerm.writeBoolean($indexTerm, $fieldValTerm)"
- case BINARY | VARBINARY =>
+ case BINARY | VARBINARY | UUID =>
s"$writerTerm.writeBinary($indexTerm, $fieldValTerm)"
case DECIMAL =>
s"$writerTerm.writeDecimal($indexTerm, $fieldValTerm,
${getPrecision(t)})"
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
index 3dc12b93000..3645c7b46c1 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
@@ -146,8 +146,8 @@ class ExpressionReducer(
unreduced.getType.getSqlTypeName match {
// we insert the original expression for object literals
case SqlTypeName.ANY | SqlTypeName.OTHER | SqlTypeName.ROW |
SqlTypeName.STRUCTURED |
- SqlTypeName.ARRAY | SqlTypeName.MAP | SqlTypeName.MULTISET |
- SqlTypeName.VARIANT =>
+ SqlTypeName.ARRAY | SqlTypeName.MAP | SqlTypeName.MULTISET |
SqlTypeName.VARIANT |
+ SqlTypeName.UUID =>
reducedValues.add(unreduced)
case SqlTypeName.VARCHAR | SqlTypeName.CHAR =>
val escapeVarchar = BinaryStringDataUtil.safeToString(
@@ -274,7 +274,7 @@ class ExpressionReducer(
// we don't support object literals yet, we skip those constant
expressions
case (SqlTypeName.ANY, _) | (SqlTypeName.OTHER, _) | (SqlTypeName.ROW,
_) |
(SqlTypeName.STRUCTURED, _) | (SqlTypeName.ARRAY, _) |
(SqlTypeName.MAP, _) |
- (SqlTypeName.MULTISET, _) | (SqlTypeName.VARIANT, _) =>
+ (SqlTypeName.MULTISET, _) | (SqlTypeName.VARIANT, _) |
(SqlTypeName.UUID, _) =>
None
case (_, call: RexCall) => {
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
index 8a864de361c..d8b2490ac61 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
@@ -314,7 +314,7 @@ object GenerateUtils {
// so that the literalValue can be also used directly when needed
generateNonNullLiteral(literalType, field, str)
- case BINARY | VARBINARY =>
+ case BINARY | VARBINARY | UUID =>
val bytesVal = literalValue.asInstanceOf[Array[Byte]]
val fieldTerm =
ctx.addReusableObject(bytesVal, "binary",
bytesVal.getClass.getCanonicalName)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
index c5f66f8c46e..a05ab9a01fb 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
@@ -435,7 +435,7 @@ object FlinkRelMdSize {
SqlTypeName.DATE =>
12d
case SqlTypeName.ANY | SqlTypeName.OTHER => 128d // 128 is an arbitrary
estimate
- case SqlTypeName.BINARY | SqlTypeName.VARBINARY | SqlTypeName.VARIANT =>
+ case SqlTypeName.BINARY | SqlTypeName.VARBINARY | SqlTypeName.VARIANT |
SqlTypeName.UUID =>
16d // 16 is an arbitrary estimate
case _ => throw new TableException(s"Unsupported data type encountered:
$sqlType")
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidSemanticTest.java
new file mode 100644
index 00000000000..debd1873403
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidSemanticTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.DataTypes;
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Semantic tests for the {@link DataTypes#UUID()} type. */
+public class UuidSemanticTest extends SemanticTestBase {
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(
+ UuidTestPrograms.UUID_SOURCE_SINK,
+ UuidTestPrograms.UUID_LITERAL,
+ UuidTestPrograms.UUID_ARRAY,
+ UuidTestPrograms.UUID_MAP,
+ UuidTestPrograms.UUID_NESTED_ROW,
+ UuidTestPrograms.UUID_INVALID_LITERAL);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidTestPrograms.java
new file mode 100644
index 00000000000..4e60704520c
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UuidTestPrograms.java
@@ -0,0 +1,115 @@
+/*
+ * 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.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+import java.util.Map;
+import java.util.UUID;
+
+/** {@link TableTestProgram}s for the {@link DataTypes#UUID()} type. */
+public class UuidTestPrograms {
+
+ private static final String LITERAL_A =
"550e8400-e29b-41d4-a716-446655440000";
+ private static final String LITERAL_B =
"f47ac10b-58cc-4372-a567-0e02b2c3d479";
+ private static final UUID UUID_A = UUID.fromString(LITERAL_A);
+ private static final UUID UUID_B = UUID.fromString(LITERAL_B);
+
+ private static SourceTestStep singleRowDriver() {
+ return SourceTestStep.newBuilder("t").addSchema("d
INT").producedValues(Row.of(1)).build();
+ }
+
+ static final TableTestProgram UUID_SOURCE_SINK =
+ TableTestProgram.of("uuid-source-sink", "round-trips a UUID column
including null")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema("id UUID")
+ .producedValues(Row.of(UUID_A),
Row.of(UUID_B), new Row(1))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("id UUID")
+ .consumedValues(Row.of(UUID_A),
Row.of(UUID_B), new Row(1))
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id FROM t")
+ .build();
+
+ static final TableTestProgram UUID_LITERAL =
+ TableTestProgram.of("uuid-literal", "materializes a UUID literal")
+ .setupTableSource(singleRowDriver())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("u UUID")
+ .consumedValues(Row.of(UUID_A))
+ .build())
+ .runSql("INSERT INTO sink_t SELECT UUID '" + LITERAL_A +
"' FROM t")
+ .build();
+
+ static final TableTestProgram UUID_ARRAY =
+ TableTestProgram.of("uuid-array", "reads UUID array elements")
+ .setupTableSource(singleRowDriver())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("arr ARRAY<UUID>")
+ .consumedValues(Row.of((Object) new UUID[]
{UUID_A, UUID_B}))
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT ARRAY[UUID '"
+ + LITERAL_A
+ + "', UUID '"
+ + LITERAL_B
+ + "'] FROM t")
+ .build();
+
+ static final TableTestProgram UUID_MAP =
+ TableTestProgram.of("uuid-map", "reads a UUID map value")
+ .setupTableSource(singleRowDriver())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("m MAP<STRING, UUID>")
+ .consumedValues(Row.of(Map.of("a",
UUID_A)))
+ .build())
+ .runSql("INSERT INTO sink_t SELECT MAP['a', UUID '" +
LITERAL_A + "'] FROM t")
+ .build();
+
+ static final TableTestProgram UUID_NESTED_ROW =
+ TableTestProgram.of("uuid-nested-row", "reads a UUID field of a
nested row")
+ .setupTableSource(singleRowDriver())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("r ROW<f0 UUID, f1 INT>")
+ .consumedValues(Row.of(Row.of(UUID_A, 42)))
+ .build())
+ .runSql("INSERT INTO sink_t SELECT (UUID '" + LITERAL_A +
"', 42) FROM t")
+ .build();
+
+ static final TableTestProgram UUID_INVALID_LITERAL =
+ TableTestProgram.of("uuid-invalid-literal", "rejects a malformed
UUID literal")
+ .setupTableSource(singleRowDriver())
+ .runFailingSql(
+ "SELECT UUID 'abcd' FROM t",
+ ValidationException.class,
+ "Invalid UUID string: abcd")
+ .build();
+}
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java
index f01a1aca779..9d7ffdaea3d 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java
@@ -41,6 +41,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
import java.util.function.Supplier;
/**
@@ -197,6 +198,8 @@ public final class DataStructureConverters {
putConverter(LogicalTypeRoot.STRUCTURED_TYPE, RowData.class,
identity());
putConverter(LogicalTypeRoot.RAW, byte[].class,
RawByteArrayConverter::create);
putConverter(LogicalTypeRoot.RAW, RawValueData.class, identity());
+ putConverter(LogicalTypeRoot.UUID, UUID.class,
constructor(UuidUuidConverter::new));
+ putConverter(LogicalTypeRoot.UUID, byte[].class, identity());
putConverter(LogicalTypeRoot.VARIANT, Variant.class, identity());
putConverter(LogicalTypeRoot.BITMAP, Bitmap.class,
constructor(BitmapBitmapConverter::new));
putConverter(LogicalTypeRoot.BITMAP, RoaringBitmapData.class,
identity());
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
new file mode 100644
index 00000000000..a3f97f9da6c
--- /dev/null
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/UuidUuidConverter.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.data.conversion;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.types.logical.UuidType;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+/**
+ * Converter for {@link UuidType} of {@link UUID} external type.
+ *
+ * <p>The internal representation is the canonical 16-byte big-endian
encoding: the most significant
+ * 64 bits followed by the least significant 64 bits.
+ */
+@Internal
+public class UuidUuidConverter implements DataStructureConverter<byte[], UUID>
{
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public byte[] toInternal(UUID external) {
+ return ByteBuffer.allocate(16)
+ .putLong(external.getMostSignificantBits())
+ .putLong(external.getLeastSignificantBits())
+ .array();
+ }
+
+ @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);
+ }
+}
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java
index 857c2ecd424..0bfebd4cb99 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryArrayWriter.java
@@ -247,6 +247,7 @@ public final class BinaryArrayWriter extends
AbstractBinaryWriter {
case RAW:
case VARIANT:
case BITMAP:
+ case UUID:
return BinaryArrayWriter::setNullLong;
case BOOLEAN:
return BinaryArrayWriter::setNullBoolean;
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java
index 4db8977ccd2..13efc76cf56 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/writer/BinaryWriter.java
@@ -166,6 +166,7 @@ public interface BinaryWriter {
break;
case BINARY:
case VARBINARY:
+ case UUID:
writer.writeBinary(pos, (byte[]) o);
break;
case VARIANT:
@@ -194,6 +195,7 @@ public interface BinaryWriter {
return (writer, pos, value) -> writer.writeBoolean(pos,
(boolean) value);
case BINARY:
case VARBINARY:
+ case UUID:
return (writer, pos, value) -> writer.writeBinary(pos,
(byte[]) value);
case DECIMAL:
final int decimalPrecision = getPrecision(elementType);
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
index 4334591cbd1..ef4d7477634 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
@@ -75,6 +75,7 @@ public final class InternalSerializers {
return BooleanSerializer.INSTANCE;
case BINARY:
case VARBINARY:
+ case UUID:
return BytePrimitiveArraySerializer.INSTANCE;
case DECIMAL:
return new DecimalDataSerializer(getPrecision(type),
getScale(type));
diff --git
a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java
index 2c19c579a4e..8336f90a182 100644
---
a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java
+++
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java
@@ -57,6 +57,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
import java.util.function.Supplier;
import static java.util.Arrays.asList;
@@ -108,6 +109,10 @@ class DataStructureConvertersTest {
.convertedTo(byte[].class, new byte[] {1, 2, 3, 4, 5}),
TestSpec.forDataType(VARBINARY(100))
.convertedTo(byte[].class, new byte[] {1, 2, 3, 4, 5}),
+ TestSpec.forDataType(DataTypes.UUID())
+ .convertedTo(
+ UUID.class,
+
UUID.fromString("550e8400-e29b-41d4-a716-446655440000")),
TestSpec.forDataType(DECIMAL(3, 2))
.convertedTo(BigDecimal.class, new BigDecimal("1.23"))
.convertedTo(DecimalData.class,
DecimalData.fromUnscaledLong(123, 3, 2)),
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
index b46b69ddcc6..7f91776f841 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
@@ -31,6 +31,7 @@ import
org.apache.flink.api.common.typeutils.base.LocalDateTimeSerializer;
import org.apache.flink.api.common.typeutils.base.LocalTimeSerializer;
import org.apache.flink.api.common.typeutils.base.NullValueSerializer;
import org.apache.flink.api.common.typeutils.base.SetSerializer;
+import org.apache.flink.api.common.typeutils.base.UuidSerializer;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import
org.apache.flink.api.common.typeutils.base.array.BooleanPrimitiveArraySerializer;
import
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
@@ -269,7 +270,8 @@ class TypeSerializerTestCoverageTest {
RowSqnInfoSerializer.class.getName(),
MetaSqnInfoSerializer.class.getName(),
SetSerializer.class.getName(),
- SortedLongSerializer.class.getName());
+ SortedLongSerializer.class.getName(),
+ UuidSerializer.class.getName());
// check if a test exists for each type serializer
for (Class<? extends TypeSerializer> typeSerializer : typeSerializers)
{
