[FLINK-40488][table] Support UUID as a comparison, ordering key

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java

@@ -58,6 +58,7 @@
 import java.util.Locale;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -306,6 +307,8 @@ public String asSerializableString(SqlFactory sqlFactory) {
                                 .map(EncodingUtils::escapeBackticks)
                                 .map(c -> String.format("`%s`", c))
                                 .collect(Collectors.joining()));
+            case UUID:
+                return String.format("UUID '%s'", getValueAs(UUID.class).get());
             case ARRAY:
             case MULTISET:
             case MAP:

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/BitmapType.java

@@ -20,11 +20,9 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.types.bitmap.Bitmap;
-import org.apache.flink.types.bitmap.RoaringBitmapData;
 
 import java.util.Collections;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Data type of bitmap data.
@@ -39,9 +37,6 @@ public final class BitmapType extends LogicalType {
 
     private static final long serialVersionUID = 1L;
 
-    private static final Set<String> INPUT_OUTPUT_CONVERSION =
-            conversionSet(Bitmap.class.getName(), RoaringBitmapData.class.getName());
-
     public BitmapType(boolean isNullable) {
         super(isNullable, LogicalTypeRoot.BITMAP);
     }
@@ -62,12 +57,12 @@ public String asSerializableString() {
 
     @Override
     public boolean supportsInputConversion(Class<?> clazz) {
-        return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+        return Bitmap.class.isAssignableFrom(clazz);
     }
 
     @Override
     public boolean supportsOutputConversion(Class<?> clazz) {
-        return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+        return Bitmap.class.isAssignableFrom(clazz);
     }
 
     @Override

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/VariantType.java

@@ -23,7 +23,6 @@
 
 import java.util.Collections;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Data type of semi-structured data.
@@ -38,8 +37,7 @@
 @PublicEvolving
 public final class VariantType extends LogicalType {
 
-    private static final Set<String> INPUT_OUTPUT_CONVERSION =
-            conversionSet(Variant.class.getName());
+    private static final long serialVersionUID = 1L;
 
     public VariantType(boolean isNullable) {
         super(isNullable, LogicalTypeRoot.VARIANT);
@@ -61,12 +59,12 @@ public String asSerializableString() {
 
     @Override
     public boolean supportsInputConversion(Class<?> clazz) {
-        return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+        return Variant.class.isAssignableFrom(clazz);
     }
 
     @Override
     public boolean supportsOutputConversion(Class<?> clazz) {
-        return INPUT_OUTPUT_CONVERSION.contains(clazz.getName());
+        return Variant.class.isAssignableFrom(clazz);
     }
 
     @Override

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ClassDataTypeConverter.java

@@ -29,7 +29,6 @@
 import org.apache.flink.types.ColumnList;
 import org.apache.flink.types.Row;
 import org.apache.flink.types.bitmap.Bitmap;
-import org.apache.flink.types.bitmap.RoaringBitmapData;
 import org.apache.flink.types.variant.Variant;
 
 import java.math.BigDecimal;
@@ -80,9 +79,6 @@ public final class ClassDataTypeConverter {
                 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());
     }
 
     private static void addDefaultDataType(Class<?> clazz, DataType rootType) {
@@ -115,6 +111,14 @@ public static Optional<DataType> extractDataType(Class<?> clazz) {
             return Optional.of(new AtomicDataType(new SymbolType<>(), clazz));
         }
 
+        if (Variant.class.isAssignableFrom(clazz)) {
+            return Optional.of(DataTypes.VARIANT());
+        }
+
+        if (Bitmap.class.isAssignableFrom(clazz)) {
+            return Optional.of(DataTypes.BITMAP());
+        }
+
         return Optional.ofNullable(defaultDataTypes.get(clazz.getName()));
     }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/ValueDataTypeConverter.java

@@ -25,7 +25,7 @@
 import org.apache.flink.table.types.logical.BinaryType;
 import org.apache.flink.table.types.logical.CharType;
 import org.apache.flink.table.types.logical.LogicalTypeFamily;
-import org.apache.flink.types.bitmap.RoaringBitmapData;
+import org.apache.flink.types.bitmap.Bitmap;
 import org.apache.flink.types.variant.Variant;
 
 import java.math.BigDecimal;
@@ -91,12 +91,6 @@ else if (value instanceof byte[]) {
             // don't let the class-based extraction kick in if array elements differ
             return convertToArrayType((Object[]) value)
                     .map(dt -> dt.notNull().bridgedTo(value.getClass()));
-        } else if (value instanceof Variant) {
-            // BinaryVariant is internal, so the conversion class is the Variant interface rather
-            // than the runtime class of the value.
-            return Optional.of(DataTypes.VARIANT().notNull());
-        } else if (value instanceof RoaringBitmapData) {
-            convertedDataType = DataTypes.BITMAP();
         }
 
         final Optional<DataType> resultType;
@@ -108,7 +102,16 @@ else if (value instanceof byte[]) {
             // DATE, TIME with java.sql.Time, and arrays of primitive types
             resultType = ClassDataTypeConverter.extractDataType(value.getClass());
         }
-        return resultType.map(dt -> dt.notNull().bridgedTo(value.getClass()));
+        return resultType.map(
+                dt -> {
+                    final DataType notNullDataType = dt.notNull();
+                    // Because they are interfaces, and we want to avoid bridgeTo internal
+                    // conversion classes.
+                    if (value instanceof Variant || value instanceof Bitmap) {
+                        return notNullDataType;
+                    }
+                    return notNullDataType.bridgedTo(value.getClass());
+                });
     }
 
     private static DataType convertToCharType(String string) {

flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java

@@ -47,6 +47,7 @@
 import java.time.temporal.ChronoField;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.UUID;
 import java.util.stream.Stream;
 
 import static java.util.Arrays.asList;
@@ -244,6 +245,16 @@ void testInstantValueLiteralExtraction() {
                 .isEqualTo(instant.minusMillis(100));
     }
 
+    @Test
+    void testUuidValueLiteralExtraction() {
+        final UUID uuid = UUID.fromString("550e8400-e29b-41d4-a716-446655440000");
+        assertThat(
+                        new ValueLiteralExpression(uuid)
+                                .getValueAs(UUID.class)
+                                .orElseThrow(AssertionError::new))
+                .isEqualTo(uuid);
+    }
+
     @Test
     void testOffsetDateTimeValueLiteralExtraction() {
         final OffsetDateTime offsetDateTime =

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ClassDataTypeConverterTest.java

@@ -24,7 +24,6 @@
 import org.apache.flink.table.types.utils.ClassDataTypeConverter;
 import org.apache.flink.types.Row;
 import org.apache.flink.types.bitmap.Bitmap;
-import org.apache.flink.types.bitmap.RoaringBitmapData;
 import org.apache.flink.types.variant.Variant;
 
 import org.junit.jupiter.params.ParameterizedTest;
@@ -96,8 +95,7 @@ private static Stream<Arguments> testData() {
                 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)));
+                of(Bitmap.class, DataTypes.BITMAP()));
     }
 
     @ParameterizedTest(name = "[{index}] class: {0} type: {1}")

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java

@@ -307,6 +307,9 @@ private static Stream<TestSpec> testData() {
                         .expectUnresolvedString("['EnumTypeInfo<java.time.DayOfWeek>']")
                         .lookupReturns(dummyRaw(DayOfWeek.class))
                         .expectResolvedDataType(dummyRaw(DayOfWeek.class)),
+                TestSpec.forUnresolvedDataType(DataTypes.of(UUID.class))
+                        .expectUnresolvedString("['java.util.UUID']")
+                        .expectResolvedDataType(UUID()),
                 TestSpec.forUnresolvedDataType(DataTypes.of(Variant.class))
                         .expectUnresolvedString("['org.apache.flink.types.variant.Variant']")
                         .expectResolvedDataType(VARIANT()),

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/ValueDataTypeConverterTest.java

@@ -25,7 +25,6 @@
 import org.apache.flink.table.types.logical.SymbolType;
 import org.apache.flink.table.types.utils.ValueDataTypeConverter;
 import org.apache.flink.types.bitmap.Bitmap;
-import org.apache.flink.types.bitmap.RoaringBitmapData;
 import org.apache.flink.types.variant.Variant;
 
 import org.junit.jupiter.params.ParameterizedTest;
@@ -43,6 +42,7 @@
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.UUID;
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -121,10 +121,9 @@ private static Stream<Arguments> testData() {
                 of(TimePointUnit.HOUR, new AtomicDataType(new SymbolType<>(), TimePointUnit.class)),
                 of(new BigDecimal[0], null),
                 of(Variant.newBuilder().of("hello"), DataTypes.VARIANT()),
-                of(Bitmap.empty(), DataTypes.BITMAP().bridgedTo(RoaringBitmapData.class)),
-                of(
-                        Bitmap.fromArray(new int[] {1, 2}),
-                        DataTypes.BITMAP().bridgedTo(RoaringBitmapData.class)));
+                of(Bitmap.empty(), DataTypes.BITMAP()),
+                of(Bitmap.fromArray(new int[] {1, 2}), DataTypes.BITMAP()),
+                of(UUID.fromString("550e8400-e29b-41d4-a716-446655440000"), DataTypes.UUID()));
     }
 
     @ParameterizedTest(name = "[{index}] value: {0} type: {1}")

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java

@@ -883,14 +883,6 @@ private static Stream<TestSpec> functionSpecs() {
                                 StaticArgument.scalar("bitmap", DataTypes.BITMAP(), false))
                         .expectAccumulator(TypeStrategies.explicit(DataTypes.BITMAP()))
                         .expectOutput(TypeStrategies.explicit(DataTypes.BITMAP())),
-                TestSpec.forScalarFunction(
-                                "Bitmap bridged to custom Bitmap",
-                                InvalidCustomBitmapTypeFunction1.class)
-                        .expectErrorMessage(
-                                "Logical type 'BITMAP' does not support a conversion from or to class 'org.apache.flink.table.types.extraction.TypeInferenceExtractorTest$CustomBitmap'."),
-                TestSpec.forScalarFunction("Custom Bitmap", InvalidCustomBitmapTypeFunction2.class)
-                        .expectErrorMessage(
-                                "Could not extract a valid type inference for function class 'org.apache.flink.table.types.extraction.TypeInferenceExtractorTest$InvalidCustomBitmapTypeFunction2'."),
                 // ---
                 TestSpec.forScalarFunction("Variant in scalar function", VariantTypeFunction.class)
                         .expectStaticArgument(
@@ -2752,77 +2744,4 @@ public Variant eval(
             return null;
         }
     }
-
-    @FunctionHint(input = @DataTypeHint(value = "BITMAP", bridgedTo = CustomBitmap.class))
-    private static class InvalidCustomBitmapTypeFunction1 extends ScalarFunction {
-        public Bitmap eval(Bitmap bitmap) {
-            return null;
-        }
-    }
-
-    private static class InvalidCustomBitmapTypeFunction2 extends ScalarFunction {
-        public Bitmap eval(CustomBitmap bitmap) {
-            return null;
-        }
-    }
-
-    public static class CustomBitmap implements Bitmap {
-
-        @Override
-        public void add(int value) {}
-
-        @Override
-        public void add(long rangeStart, long rangeEnd) {}
-
-        @Override
-        public void addN(int[] values, int offset, int n) {}
-
-        @Override
-        public void and(@Nullable Bitmap other) {}
-
-        @Override
-        public void andNot(@Nullable Bitmap other) {}
-
-        @Override
-        public void clear() {}
-
-        @Override
-        public boolean contains(int value) {
-            return false;
-        }
-
-        @Override
-        public int getCardinality() {
-            return 0;
-        }
-
-        @Override
-        public long getLongCardinality() {
-            return 0;
-        }
-
-        @Override
-        public boolean isEmpty() {
-            return false;
-        }
-
-        @Override
-        public void or(@Nullable Bitmap other) {}
-
-        @Override
-        public void remove(int value) {}
-
-        @Override
-        public int[] toArray() {
-            return new int[0];
-        }
-
-        @Override
-        public byte[] toBytes() {
-            return new byte[0];
-        }
-
-        @Override
-        public void xor(@Nullable Bitmap other) {}
-    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java

@@ -74,6 +74,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.table.planner.typeutils.SymbolUtil.commonToCalcite;
@@ -141,6 +142,13 @@ public RexNode visit(ValueLiteralExpression valueLiteral) {
                             .collect(Collectors.toList()));
         }
 
+        if (type.is(LogicalTypeRoot.UUID)) {
+            // UUID has no generic RexBuilder#makeLiteral support, so build the literal directly.
+            // This also lets filter push-down round-trip a UUID predicate back into a RexNode.
+            return rexBuilder.makeUuidLiteral(
+                    valueLiteral.getValueAs(UUID.class).orElseThrow(IllegalStateException::new));
+        }
+
         Object value;
         switch (type.getTypeRoot()) {
             case DECIMAL:

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

@@ -629,7 +629,9 @@ object GenerateUtils {
       s"$leftTerm.compareTo($rightTerm)"
     case BOOLEAN =>
       s"($leftTerm == $rightTerm ? 0 : ($leftTerm ? 1 : -1))"
-    case BINARY | VARBINARY =>
+    case BINARY | VARBINARY | UUID =>
+      // UUID is stored as its 16-byte big-endian encoding, so it orders by the same unsigned
+      // byte-wise comparison as binary strings.
       val sortUtil =
         classOf[org.apache.flink.table.runtime.operators.sort.SortUtil].getCanonicalName
       s"$sortUtil.compareBinary($leftTerm, $rightTerm)"

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala

@@ -654,9 +654,10 @@ object ScalarOperatorGens {
           case _ => throw new CodeGenException(s"Unsupported boolean comparison '$operator'.")
         }
       }
-      // both sides are binary type
+      // both sides are binary type or UUID (both are backed by a byte[] and order by the same
+      // unsigned byte-wise comparison)
       else if (
-        isBinaryString(left.resultType) &&
+        (isBinaryString(left.resultType) || isUuid(left.resultType)) &&
         isInteroperable(left.resultType, right.resultType)
       ) {
         val utilName = classOf[SqlFunctionUtils].getCanonicalName