This is an automated email from the ASF dual-hosted git repository.
gustavodemorais 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 878844f937f [FLINK-40486][table] Introduce the UUID logical type
878844f937f is described below
commit 878844f937f78963b8765d848731d13b4592ce32
Author: Ramin Gharib <[email protected]>
AuthorDate: Tue Sep 1 10:48:40 2026 +0200
[FLINK-40486][table] Introduce the UUID logical type
This closes #29041
---
docs/static/generated/rest_v1_sql_gateway.yml | 1 +
docs/static/generated/rest_v2_sql_gateway.yml | 1 +
docs/static/generated/rest_v3_sql_gateway.yml | 1 +
docs/static/generated/rest_v4_sql_gateway.yml | 1 +
.../rest/serde/LogicalTypeJsonDeserializer.java | 3 +
.../rest/serde/LogicalTypeJsonSerializer.java | 1 +
.../rest/serde/LogicalTypeJsonSerDeTest.java | 2 +
.../java/org/apache/flink/table/api/DataTypes.java | 12 +++
.../flink/table/types/logical/LogicalTypeRoot.java | 4 +-
.../table/types/logical/LogicalTypeVisitor.java | 4 +
.../apache/flink/table/types/logical/UuidType.java | 85 ++++++++++++++++++++++
.../types/logical/utils/LogicalTypeParser.java | 6 +-
.../apache/flink/table/types/DataTypesTest.java | 6 ++
.../flink/table/types/LogicalTypeParserTest.java | 3 +
.../apache/flink/table/types/LogicalTypesTest.java | 16 ++++
.../table/planner/calcite/FlinkTypeFactory.java | 7 ++
.../exec/serde/LogicalTypeJsonSerializer.java | 2 +
.../typeutils/LogicalRelDataTypeConverter.java | 8 ++
.../planner/calcite/FlinkTypeFactoryTest.java | 2 +
.../nodes/exec/serde/LogicalTypeJsonSerdeTest.java | 2 +
.../typeutils/LogicalRelDataTypeConverterTest.java | 2 +
.../table/runtime/typeutils/TypeCheckUtils.java | 8 +-
22 files changed, 174 insertions(+), 3 deletions(-)
diff --git a/docs/static/generated/rest_v1_sql_gateway.yml
b/docs/static/generated/rest_v1_sql_gateway.yml
index fecf0576a01..8a7d1688747 100644
--- a/docs/static/generated/rest_v1_sql_gateway.yml
+++ b/docs/static/generated/rest_v1_sql_gateway.yml
@@ -403,6 +403,7 @@ components:
- DESCRIPTOR
- VARIANT
- BITMAP
+ - UUID
OpenSessionRequestBody:
type: object
properties:
diff --git a/docs/static/generated/rest_v2_sql_gateway.yml
b/docs/static/generated/rest_v2_sql_gateway.yml
index e39feaaea8e..51faa9c28ae 100644
--- a/docs/static/generated/rest_v2_sql_gateway.yml
+++ b/docs/static/generated/rest_v2_sql_gateway.yml
@@ -477,6 +477,7 @@ components:
- DESCRIPTOR
- VARIANT
- BITMAP
+ - UUID
OpenSessionRequestBody:
type: object
properties:
diff --git a/docs/static/generated/rest_v3_sql_gateway.yml
b/docs/static/generated/rest_v3_sql_gateway.yml
index 6d171ac2152..dd6a5ca1a5f 100644
--- a/docs/static/generated/rest_v3_sql_gateway.yml
+++ b/docs/static/generated/rest_v3_sql_gateway.yml
@@ -506,6 +506,7 @@ components:
- DESCRIPTOR
- VARIANT
- BITMAP
+ - UUID
OpenSessionRequestBody:
type: object
properties:
diff --git a/docs/static/generated/rest_v4_sql_gateway.yml
b/docs/static/generated/rest_v4_sql_gateway.yml
index d39da5e007c..cd7061c1b3c 100644
--- a/docs/static/generated/rest_v4_sql_gateway.yml
+++ b/docs/static/generated/rest_v4_sql_gateway.yml
@@ -516,6 +516,7 @@ components:
- DESCRIPTOR
- VARIANT
- BITMAP
+ - UUID
OpenSessionRequestBody:
type: object
properties:
diff --git
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
index b154c81e815..8c3a88b8fee 100644
---
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
+++
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
@@ -46,6 +46,7 @@ import org.apache.flink.table.types.logical.SmallIntType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -165,6 +166,8 @@ public final class LogicalTypeJsonDeserializer extends
StdDeserializer<LogicalTy
return new VariantType(isNullable);
case BITMAP:
return new BitmapType(isNullable);
+ case UUID:
+ return new UuidType(isNullable);
default:
throw new UnsupportedOperationException(
String.format(
diff --git
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
index c7e9b3b2eb9..bb650453c4f 100644
---
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
+++
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
@@ -130,6 +130,7 @@ public final class LogicalTypeJsonSerializer extends
StdSerializer<LogicalType>
case DATE:
case VARIANT:
case BITMAP:
+ case UUID:
break;
case CHAR:
jsonGenerator.writeNumberField(
diff --git
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
index b71276fc41c..dd1f71b255f 100644
---
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
+++
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
@@ -48,6 +48,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.YearMonthIntervalType;
@@ -197,6 +198,7 @@ class LogicalTypeJsonSerDeTest {
new MultisetType(BinaryType.ofEmptyLiteral()),
new MultisetType(VarBinaryType.ofEmptyLiteral()),
new BitmapType(),
+ new UuidType(),
RowType.of(new BigIntType(), new IntType(false), new
VarCharType(200)),
RowType.of(
new LogicalType[] {
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
index af1e718d90e..9beae7d2f4a 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
@@ -59,6 +59,7 @@ import
org.apache.flink.table.types.logical.StructuredType.StructuredAttribute;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -1075,6 +1076,17 @@ public final class DataTypes {
return new AtomicDataType(new BitmapType());
}
+ /**
+ * Data type of a universally unique identifier (UUID).
+ *
+ * <p>The type represents a 128-bit value stored as a fixed 16-byte
big-endian sequence.
+ *
+ * @see UuidType
+ */
+ public static DataType UUID() {
+ return new AtomicDataType(new UuidType());
+ }
+
//
--------------------------------------------------------------------------------------------
// Helper functions
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
index 6c823add433..1026070c710 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
@@ -145,7 +145,9 @@ public enum LogicalTypeRoot {
VARIANT(LogicalTypeFamily.EXTENSION),
- BITMAP(LogicalTypeFamily.EXTENSION);
+ BITMAP(LogicalTypeFamily.EXTENSION),
+
+ UUID(LogicalTypeFamily.EXTENSION);
private final Set<LogicalTypeFamily> families;
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
index 6a0e5614466..fe5548c8f65 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
@@ -101,5 +101,9 @@ public interface LogicalTypeVisitor<R> {
return visit((LogicalType) bitmapType);
}
+ default R visit(UuidType uuidType) {
+ return visit((LogicalType) uuidType);
+ }
+
R visit(LogicalType other);
}
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
new file mode 100644
index 00000000000..2973a89b601
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/UuidType.java
@@ -0,0 +1,85 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Data type of a universally unique identifier (UUID).
+ *
+ * <p>The type stores any 128-bit UUID as the canonical 16-byte big-endian
encoding defined by RFC
+ * 9562 (which obsoletes RFC 4122). It is version-agnostic: the version and
variant bits are stored
+ * as-is and never validated or interpreted.
+ *
+ * <p>The serializable string representation of this type is {@code UUID}.
+ */
+@PublicEvolving
+public final class UuidType extends LogicalType {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Set<String> INPUT_OUTPUT_CONVERSION =
conversionSet(UUID.class.getName());
+
+ public UuidType(boolean isNullable) {
+ super(isNullable, LogicalTypeRoot.UUID);
+ }
+
+ public UuidType() {
+ this(true);
+ }
+
+ @Override
+ public LogicalType copy(boolean isNullable) {
+ return new UuidType(isNullable);
+ }
+
+ @Override
+ public String asSerializableString() {
+ return withNullability("UUID");
+ }
+
+ @Override
+ public boolean supportsInputConversion(Class<?> clazz) {
+ return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+ }
+
+ @Override
+ public boolean supportsOutputConversion(Class<?> clazz) {
+ return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+ }
+
+ @Override
+ public Class<?> getDefaultConversion() {
+ return UUID.class;
+ }
+
+ @Override
+ public List<LogicalType> getChildren() {
+ return List.of();
+ }
+
+ @Override
+ public <R> R accept(LogicalTypeVisitor<R> visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
index 3711ac76f4f..a8e85920433 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
@@ -54,6 +54,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.UnresolvedUserDefinedType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -336,7 +337,8 @@ public final class LogicalTypeParser {
DESCRIPTOR,
STRUCTURED,
VARIANT,
- BITMAP
+ BITMAP,
+ UUID
}
private static final Set<String> KEYWORDS =
@@ -586,6 +588,8 @@ public final class LogicalTypeParser {
return new VariantType();
case BITMAP:
return new BitmapType();
+ case UUID:
+ return new UuidType();
default:
throw parsingError("Unsupported type: " + token().value);
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
index 8c274ed7191..7151f9b8c22 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
@@ -44,6 +44,7 @@ import org.apache.flink.table.types.logical.SmallIntType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -68,6 +69,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.UUID;
import java.util.stream.Stream;
import static org.apache.flink.table.api.DataTypes.ARRAY;
@@ -100,6 +102,7 @@ import static
org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
import static
org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE;
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.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
import static org.apache.flink.table.api.DataTypes.VARIANT;
@@ -231,6 +234,9 @@ class DataTypesTest {
TestSpec.forDataType(BITMAP())
.expectLogicalType(new BitmapType())
.expectConversionClass(Bitmap.class),
+ TestSpec.forDataType(UUID())
+ .expectLogicalType(new UuidType())
+ .expectConversionClass(UUID.class),
TestSpec.forUnresolvedDataType(RAW(Types.VOID))
.expectUnresolvedString("[RAW('java.lang.Void', '?')]")
.lookupReturns(dummyRaw(Void.class))
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
index 258b397c871..8a87c4663f4 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
@@ -55,6 +55,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.UnresolvedUserDefinedType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -309,6 +310,8 @@ public class LogicalTypeParserTest {
TestSpec.forString("VARIANT NOT NULL").expectType(new
VariantType(false)),
TestSpec.forString("BITMAP").expectType(new BitmapType()),
TestSpec.forString("BITMAP NOT NULL").expectType(new
BitmapType(false)),
+ TestSpec.forString("UUID").expectType(new UuidType()),
+ TestSpec.forString("UUID NOT NULL").expectType(new
UuidType(false)),
// error message testing
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 8b06ceb5118..0703e16c78e 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
@@ -57,6 +57,7 @@ import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.UnresolvedUserDefinedType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -79,6 +80,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import java.util.UUID;
import static org.apache.flink.table.test.TableAssertions.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
@@ -626,6 +628,20 @@ public class LogicalTypesTest {
new BitmapType(false)));
}
+ @Test
+ void testUuidType() {
+ assertThat(new UuidType())
+ .isJavaSerializable()
+ .satisfies(
+ baseAssertions(
+ "UUID",
+ "UUID",
+ new Class[] {UUID.class},
+ new Class[] {UUID.class},
+ new LogicalType[] {},
+ new UuidType(false)));
+ }
+
@Test
void testTypeInformationRawType() {
final TypeInformationRawType<?> rawType =
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java
index f5bd64adeef..4157e378da8 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkTypeFactory.java
@@ -63,6 +63,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -490,6 +491,9 @@ public class FlinkTypeFactory extends JavaTypeFactoryImpl
implements ExtendedRel
case VARIANT:
return createSqlType(SqlTypeName.VARIANT);
+ case UUID:
+ return createSqlType(SqlTypeName.UUID);
+
case BITMAP:
return new BitmapRelDataType((BitmapType) logicalType);
@@ -861,6 +865,9 @@ public class FlinkTypeFactory extends JavaTypeFactoryImpl
implements ExtendedRel
case VARIANT:
return new VariantType();
+ case UUID:
+ return new UuidType();
+
case OTHER:
if (relDataType instanceof RawRelDataType) {
return ((RawRelDataType) relDataType).getRawType();
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
index f06c38ba1b8..27bdff6e398 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
@@ -239,6 +239,7 @@ final class LogicalTypeJsonSerializer extends
StdSerializer<LogicalType> {
break;
case SYMBOL:
case VARIANT:
+ case UUID:
// type root is enough
break;
case RAW:
@@ -546,6 +547,7 @@ final class LogicalTypeJsonSerializer extends
StdSerializer<LogicalType> {
case DESCRIPTOR:
case BITMAP:
case VARIANT:
+ case UUID:
return true;
default:
// fall back to generic serialization
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java
index cf78ce82068..58c164b31ec 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverter.java
@@ -57,6 +57,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -467,6 +468,11 @@ public final class LogicalRelDataTypeConverter {
return relDataTypeFactory.createSqlType(SqlTypeName.VARIANT);
}
+ @Override
+ public RelDataType visit(UuidType uuidType) {
+ return relDataTypeFactory.createSqlType(SqlTypeName.UUID);
+ }
+
@Override
public RelDataType visit(BitmapType bitmapType) {
return new BitmapRelDataType(bitmapType);
@@ -596,6 +602,8 @@ public final class LogicalRelDataTypeConverter {
return new DescriptorType(false);
case VARIANT:
return new VariantType(false);
+ case UUID:
+ return new UuidType(false);
case STRUCTURED:
case OTHER:
if (relDataType instanceof StructuredRelDataType) {
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java
index 6e0e7a8c3d3..a5a85e11c3b 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkTypeFactoryTest.java
@@ -46,6 +46,7 @@ import org.apache.flink.table.types.logical.SmallIntType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
@@ -89,6 +90,7 @@ class FlinkTypeFactoryTest {
new TimeType(),
new TimestampType(3),
new LocalZonedTimestampType(3),
+ new UuidType(),
new ArrayType(new DoubleType()),
new MapType(new DoubleType(), VarCharType.STRING_TYPE),
RowType.of(new DoubleType(), VarCharType.STRING_TYPE),
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java
index c838a6ca08a..32554e7264e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerdeTest.java
@@ -59,6 +59,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -268,6 +269,7 @@ public class LogicalTypeJsonSerdeTest {
new MultisetType(BinaryType.ofEmptyLiteral()),
new MultisetType(VarBinaryType.ofEmptyLiteral()),
new BitmapType(),
+ new UuidType(),
new VariantType(),
new ArrayType(new VariantType()),
new MultisetType(new VariantType()),
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverterTest.java
index fd37601a39f..b47206e184c 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/LogicalRelDataTypeConverterTest.java
@@ -48,6 +48,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.UuidType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
@@ -153,6 +154,7 @@ public class LogicalRelDataTypeConverterTest {
new MultisetType(BinaryType.ofEmptyLiteral()),
new MultisetType(VarBinaryType.ofEmptyLiteral()),
new BitmapType(),
+ new UuidType(),
new VariantType(),
new ArrayType(new VariantType()),
new MapType(new VarCharType(5), new VariantType()),
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java
index e8fe0c0ad4e..c469696d247 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java
@@ -39,6 +39,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.VARIANT;
import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isRowtimeAttribute;
@@ -146,6 +147,10 @@ public class TypeCheckUtils {
return type.getTypeRoot() == BITMAP;
}
+ private static boolean isUuidType(LogicalType type) {
+ return type.getTypeRoot() == UUID;
+ }
+
public static boolean isComparable(LogicalType type) {
return !isRaw(type)
&& !isMap(type)
@@ -154,7 +159,8 @@ public class TypeCheckUtils {
&& !isArray(type)
&& !isStructuredType(type)
&& !isVariantType(type)
- && !isBitmapType(type);
+ && !isBitmapType(type)
+ && !isUuidType(type);
}
public static boolean isMutable(LogicalType type) {
