Parquet: Fix UUID ClassCastException

java.util.UUID cannot be cast to class java.nio.ByteBuffer

Author: ndrluis

data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java

@@ -119,7 +119,8 @@ public static List<Object> parameters() {
           optional(14, "all_nans", DoubleType.get()),
           optional(15, "some_nans", FloatType.get()),
           optional(16, "no_nans", DoubleType.get()),
-          optional(17, "some_double_nans", DoubleType.get()));
+          optional(17, "some_double_nans", DoubleType.get()),
+          optional(18, "uuid_col", Types.UUIDType.get()));
 
   private static final Types.StructType UNDERSCORE_STRUCT_FIELD_TYPE =
       Types.StructType.of(Types.NestedField.required(8, "_int_field", IntegerType.get()));
@@ -137,7 +138,8 @@ public static List<Object> parameters() {
           optional(14, "_all_nans", Types.DoubleType.get()),
           optional(15, "_some_nans", FloatType.get()),
           optional(16, "_no_nans", Types.DoubleType.get()),
-          optional(17, "_some_double_nans", Types.DoubleType.get()));
+          optional(17, "_some_double_nans", Types.DoubleType.get()),
+          optional(18, "_uuid_col", Types.UUIDType.get()));
 
   private static final Schema VARIANT_SCHEMA =
       new Schema(
@@ -157,6 +159,11 @@ public static List<Object> parameters() {
   private static final int INT_MIN_VALUE = 30;
   private static final int INT_MAX_VALUE = 79;
 
+  private static final UUID UUID_WITH_ZEROS =
+      UUID.fromString("00000000-0000-0000-0000-000000000000");
+  private static final UUID UUID_WITH_ONES =
+      UUID.fromString("11111111-1111-1111-1111-111111111111");
+
   private File orcFile = null;
   private MessageType parquetSchema = null;
   private BlockMetaData rowGroupMetadata = null;
@@ -211,6 +218,9 @@ public void createOrcInputFile() throws IOException {
         structNotNull.setField("_int_field", INT_MIN_VALUE + i);
         record.setField("_struct_not_null", structNotNull); // struct with int
 
+        record.setField(
+            "_uuid_col", (i % 3 == 0) ? UUID_WITH_ZEROS : (i % 3 == 1) ? UUID_WITH_ONES : null);
+
         appender.add(record);
       }
     }
@@ -248,6 +258,10 @@ private void createParquetInputFile() throws IOException {
       GenericRecord structNotNull = GenericRecord.create(UNDERSCORE_STRUCT_FIELD_TYPE);
       structNotNull.setField("_int_field", INT_MIN_VALUE + i);
       builder.setField("_struct_not_null", structNotNull); // struct with int
+
+      builder.setField(
+          "_uuid_col", (i % 3 == 0) ? UUID_WITH_ZEROS : (i % 3 == 1) ? UUID_WITH_ONES : null);
+
       records.add(builder);
     }
 
@@ -1063,6 +1077,18 @@ public void testVariantFieldAllNullsNotNull() throws IOException {
     }
   }
 
+  @TestTemplate
+  public void testUUIDEq() {
+    assumeThat(format).as("Only valid for Parquet").isEqualTo(FileFormat.PARQUET);
+
+    boolean shouldRead = shouldRead(equal("uuid_col", UUID_WITH_ZEROS));
+    assertThat(shouldRead).as("Should read: UUID value exists in row group").isTrue();
+
+    UUID nonExistentUuid = UUID.fromString("99999999-9999-9999-9999-999999999999");
+    boolean shouldSkip = shouldRead(equal("uuid_col", nonExistentUuid));
+    assertThat(shouldSkip).as("Should skip: UUID value does not exist in row group").isFalse();
+  }
+
   private boolean shouldRead(Expression expression) {
     return shouldRead(expression, true);
   }

parquet/src/main/java/org/apache/iceberg/parquet/ParquetConversions.java

@@ -83,6 +83,8 @@ static Function<Object, Object> converterFromParquet(
       } else if (icebergType.typeId() == Type.TypeID.DOUBLE
           && parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.FLOAT) {
         return value -> ((Float) fromParquet.apply(value)).doubleValue();
+      } else if (icebergType.typeId() == Type.TypeID.UUID) {
+        return binary -> UUIDUtil.convert(((Binary) binary).toByteBuffer());
       }
     }
 

parquet/src/test/java/org/apache/iceberg/parquet/TestDictionaryRowGroupFilter.java

@@ -111,7 +111,8 @@ public class TestDictionaryRowGroupFilter {
               14,
               "decimal_fixed",
               DecimalType.of(20, 10)), // >18 precision to enforce FIXED_LEN_BYTE_ARRAY
-          optional(15, "_nans_and_nulls", DoubleType.get()));
+          optional(15, "_nans_and_nulls", DoubleType.get()),
+          optional(16, "uuid_col", Types.UUIDType.get()));
 
   private static final Types.StructType UNDERSCORE_STRUCT_FIELD_TYPE =
       Types.StructType.of(Types.NestedField.required(9, "_int_field", IntegerType.get()));
@@ -133,7 +134,8 @@ public class TestDictionaryRowGroupFilter {
               14,
               "_decimal_fixed",
               DecimalType.of(20, 10)), // >18 precision to enforce FIXED_LEN_BYTE_ARRAY
-          optional(15, "_nans_and_nulls", DoubleType.get()));
+          optional(15, "_nans_and_nulls", DoubleType.get()),
+          optional(16, "_uuid_col", Types.UUIDType.get()));
 
   private static final String TOO_LONG_FOR_STATS;
 
@@ -153,6 +155,11 @@ public class TestDictionaryRowGroupFilter {
           .subtract(DECIMAL_MIN_VALUE)
           .divide(new BigDecimal(INT_MAX_VALUE - INT_MIN_VALUE), RoundingMode.HALF_UP);
 
+  private static final UUID UUID_WITH_ZEROS =
+      UUID.fromString("00000000-0000-0000-0000-000000000000");
+  private static final UUID UUID_WITH_ONES =
+      UUID.fromString("11111111-1111-1111-1111-111111111111");
+
   private MessageType parquetSchema = null;
   private BlockMetaData rowGroupMetadata = null;
   private DictionaryPageReadStore dictionaryStore = null;
@@ -203,6 +210,9 @@ public void createInputFile() throws IOException {
           structNotNull.put("_int_field", INT_MIN_VALUE + i);
           builder.set("_struct_not_null", structNotNull); // struct with int
 
+          builder.set(
+              "_uuid_col", (i % 3 == 0) ? UUID_WITH_ZEROS : (i % 3 == 1) ? UUID_WITH_ONES : null);
+
           appender.add(builder.build());
         }
       }
@@ -1267,6 +1277,23 @@ public void testTransformFilter() {
         .isTrue();
   }
 
+  @TestTemplate
+  public void testUUIDDictionaryFilter() {
+    assumeThat(getColumnForName(rowGroupMetadata, "_uuid_col").getEncodings())
+        .contains(Encoding.RLE_DICTIONARY);
+
+    boolean shouldReadExisting =
+        new ParquetDictionaryRowGroupFilter(SCHEMA, equal("uuid_col", UUID_WITH_ZEROS))
+            .shouldRead(parquetSchema, rowGroupMetadata, dictionaryStore);
+    assertThat(shouldReadExisting).as("Should read: Dictionary contains a matching entry").isTrue();
+
+    UUID nonExistentUUID = UUID.fromString("22222222-2222-2222-2222-222222222222");
+    boolean shouldRead =
+        new ParquetDictionaryRowGroupFilter(SCHEMA, equal("uuid_col", nonExistentUUID))
+            .shouldRead(parquetSchema, rowGroupMetadata, dictionaryStore);
+    assertThat(shouldRead).as("Should skip: UUID not found in dictionary").isFalse();
+  }
+
   private ColumnChunkMetaData getColumnForName(BlockMetaData rowGroup, String columnName) {
     ColumnPath columnPath = ColumnPath.fromDotString(columnName);
     for (ColumnChunkMetaData column : rowGroup.getColumns()) {