[FLINK-40486][table] Introduce the UUID logical type

[raminqaf]

What is the purpose of the change

This is the first sub-task of FLIP-604 (Complete VARIANT Primitive Coverage with UUID and Timestamps). It introduces UUID as a native logical type in the Table/SQL type system.

UUID represents a 128-bit value stored as a fixed 16-byte big-endian sequence, with java.util.UUID as its default conversion class. It maps to Calcite's native SqlTypeName.UUID, which is available in the Calcite 1.41 version Flink already uses, so no SQL parser or Calcite version change is required. The type mirrors VARIANT: a parameterless scalar in the EXTENSION family.

This change is scoped to the type system only. It makes the type exist, be constructible, parseable, and round-trip through the planner and plan serde. It does not make queries over UUID execute. Runtime serialization and codegen (FLINK-40490), casting (FLINK-40487), comparison and ordering (FLINK-40488), functions (FLINK-40489), VARIANT integration (FLINK-40491–40493), and documentation (FLINK-40494) follow in separate sub-tasks.

Brief change log

• Add UuidType and LogicalTypeRoot.UUID (EXTENSION family); add the default visit(UuidType) method to LogicalTypeVisitor.
• Add the DataTypes.UUID() factory and the UUID keyword to LogicalTypeParser.
• Map UuidType to and from SqlTypeName.UUID in FlinkTypeFactory and LogicalRelDataTypeConverter.
• Support UUID in the planner and SQL gateway logical-type JSON serde.

Verifying this change

This change added unit tests and can be verified as follows:

• LogicalTypesType, DataTypesTest, LogicalTypeParserTest cover the type, the DataTypes.UUID() factory, and keyword parsing (UUID / UUID NOT NULL).
• FlinkTypeFactoryTest#testInternalToRelType covers the LogicalType↔RelDataType round-trip including nullability.
• LogicalRelDataTypeConverterTest, and the planner and SQL gateway LogicalTypeJson(Ser)DeTest cover JSON serde round-trips.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes (new @PublicEvolving UuidType, LogicalTypeRoot.UUID, DataTypes.UUID())
• The serializers: yes (logical-type JSON serde for compiled plans and the SQL gateway REST; no binary/state TypeSerializer is added)
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? not documented (documentation is tracked separately in FLINK-40494)

Two things to confirm before you submit:

1. The "serializers" answer. I marked it yes because the compiled-plan JSON serde is a compatibility surface and this extends it. If you consider that question to mean only binary TypeSerializers, flip it to no. I'd keep it as yes with the clarifying note.
2. The ASF generative-AI disclosure section (new in the template). AI tooling was used here, so I left it out of the message above rather than fill it falsely. Per ASF policy you should append one of:

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 4.8)

[flinkbot]

CI report:

• 2de766f Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[spuru9]

LGTM

Need to regenerate rest docs

./mvnw package -Dgenerate-rest-docs -pl flink-docs -am -nsu -DskipTests
then commit the updated docs/static/generated/rest_v1_sql_gateway.yml.

[raminqaf]

LGTM

Need to regenerate rest docs

./mvnw package -Dgenerate-rest-docs -pl flink-docs -am -nsu -DskipTests
then commit the updated docs/static/generated/rest_v1_sql_gateway.yml.

@spuru9 Thanks. Updated the docs

[gustavodemorais]

A few things I wasn't sure about while going through this:

[gustavodemorais]

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java (around line 234/242, where VARIANT/BITMAP get explicit casts from VARBINARY/BINARY): I don't see UUID registered anywhere in LogicalTypeCasts, not even an explicit cast from VARCHAR/VARBINARY like BITMAP/VARIANT have. Is cast support also part of the codegen follow-up, or planned for this PR? Wdyt?

[gustavodemorais]

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java:149 (isComparable): nit: it excludes BITMAP and VARIANT explicitly but not UUID. I think it needs the same treatment now, even with codegen as a follow-up - otherwise ORDER BY/joins on a UUID column could crash confusingly instead of failing validation cleanly in the meantime. Make sense?

[gustavodemorais]

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java:83-84: nit: it registers Bitmap.class and Variant.class but I don't see a UUID.class -> DataTypes.UUID() entry. Might be worth adding for consistency, unless there's a reason to leave it out?

[raminqaf]

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java (around line 234/242, where VARIANT/BITMAP get explicit casts from VARBINARY/BINARY): I don't see UUID registered anywhere in LogicalTypeCasts, not even an explicit cast from VARCHAR/VARBINARY like BITMAP/VARIANT have. Is cast support also part of the codegen follow-up, or planned for this PR? Wdyt?

Casting is intentionally out of this PR. It's a dedicated FLIP-604 sub-task: FLINK-40487

[raminqaf]

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/TypeCheckUtils.java:149 (isComparable): nit: it excludes BITMAP and VARIANT explicitly but not UUID. I think it needs the same treatment now, even with codegen as a follow-up - otherwise ORDER BY/joins on a UUID column could crash confusingly instead of failing validation cleanly in the meantime. Make sense?

Good catch, agreed. isComparable returns true for UUID today, so ORDER BY, joins, and comparisons on a UUID column pass validation and then hit codegen, which has no UUID support yet. Added it to TypeCheckUtils

[raminqaf]

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java:83-84: nit: it registers Bitmap.class and Variant.class but I don't see a UUID.class -> DataTypes.UUID() entry. Might be worth adding for consistency, unless there's a reason to leave it out?

My plan is to add this entry together with FLINK-40490 so the reflective path is functional the moment java.util.UUID starts resolving to UUID. If you would rather keep parity with Bitmap/Variant now, I can add it, with that caveat.

[dylanhz]

Hi, thanks for the PR!
Just a small question: FLIP-604 also lists byte[] and canonical String as supported conversion classes, while UuidType currently only supports java.util.UUID. Which PR is planned to add support for these two conversion classes?

[gustavodemorais]

My plan is to add this entry together with FLINK-40490 so the reflective path is functional the moment java.util.UUID starts resolving to UUID. If you would rather keep parity with Bitmap/Variant now, I can add it, with that caveat.

Up to you.

Btw, I don't think you pushed the changes to TypeCheckUtils

[gustavodemorais]

One last nit: this reads too complicated for what it actually does

private static final Set<String> INPUT_OUTPUT_CONVERSION = conversionSet(UUID.class.getName());

@Override
    public boolean supportsInputConversion(Class<?> clazz) {
        return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
    }

Can we simplify it? One of the recent types (DescriptorType) does a similar thing

private static final Class<?> INPUT_OUTPUT_CONVERSION = UUID.class;

@Override
    public boolean supportsInputConversion(Class<?> clazz) {
        return INPUT_OUTPUT_CONVERSION == clazz;
    }
...

Same for supportsOutputConversion

[raminqaf]

I can change this now. But on the follow-up PR FLINK-40490 we will add byte[]/String and this will go back to a set

[gustavodemorais]

Makes sense. Sounds fine to leave as-is

[snuyanzin]

there is a more critical downside of usage of classes: it only works if classloader is same
in case of different classloader == will return false
there is no such issue for classnames

[raminqaf]

Hi, thanks for the PR! Just a small question: FLIP-604 also lists byte[] and canonical String as supported conversion classes, while UuidType currently only supports java.util.UUID. Which PR is planned to add support for these two conversion classes?

Thanks @dylanhz! byte[] (16 bytes) and the canonical String are alternative conversion classes, not the default. They only make sense together with the DataStructureConverters that translate between the internal 16-byte value and those external classes, which is runtime work. A conversion class without its converter would pass supportsConversion but then fail at planning when the converter lookup comes up empty, so they can't be advertised on their own.

They land with FLINK-40490 "Runtime serialization and codegen for UUID", alongside the UUID DataStructureConverter. This PR keeps only the default java.util.UUID conversion on purpose, so the type never advertises a bridging class that has no working converter yet.

[gustavodemorais]

LGTM, thanks @raminqaf. Can you make sure CI is green?

[snuyanzin]

RFC describes multiple versions

what version we are talking bout here?

[raminqaf]

Good point. The type is version-agnostic. It stores any 128-bit UUID in the canonical 16-byte big-endian layout defined by RFC 9562 (which obsoletes RFC 4122), and it does not inspect or validate the version/variant bits, so v1, v4, v7, and so on all map to the same UUID type. Versions are only relevant to generation, which lands separately as UUID_V4() / UUID_V7() in FLINK-40489. I will reword the javadoc to say this explicitly so it does not read as tied to one version.