linliu-code commented on code in PR #13990:
URL: https://github.com/apache/hudi/pull/13990#discussion_r2379727783
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
+ case STRING:
+ return "test_string_" + RANDOM.nextInt(1000);
+ case BYTES:
+ byte[] bytes = new byte[RANDOM.nextInt(10) + 1];
+ RANDOM.nextBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ case RECORD:
+ return generateRandomRecord(actualSchema);
+ case ENUM:
+ List<String> symbols = actualSchema.getEnumSymbols();
+ return new GenericData.EnumSymbol(actualSchema,
symbols.get(RANDOM.nextInt(symbols.size())));
+ case ARRAY:
+ List<Object> array = new ArrayList<>();
+ int arraySize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < arraySize; i++) {
+ array.add(generateRandomValue(actualSchema.getElementType(),
null));
+ }
+ return array;
+ case MAP:
+ Map<String, Object> map = new HashMap<>();
+ int mapSize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < mapSize; i++) {
+ map.put("key_" + i,
generateRandomValue(actualSchema.getValueType(), null));
+ }
+ return map;
+ case FIXED:
+ byte[] fixedBytes = new byte[actualSchema.getFixedSize()];
+ RANDOM.nextBytes(fixedBytes);
+ return new GenericData.Fixed(actualSchema, fixedBytes);
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Compare a SpecificRecord with a GenericRecord using the same schema
+ */
+ public static void assertRecordsEqual(SpecificRecord specificRecord,
GenericRecord genericRecord, Schema schema) {
+ assertEquals(specificRecord.getClass().getSimpleName(),
genericRecord.getSchema().getName());
+
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values - treat JsonProperties$Null as null
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue == null || genericValue == null) {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " has different null values");
+ continue;
+ }
+ // Handle different types of comparisons based on field schema
+ Schema fieldSchema;
+ try {
+ fieldSchema = resolveNullableSchema(field.schema());
+ } catch (Exception e) {
+ // For complex union types that can't be resolved, do content-based
comparison
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ // Compare wrapper types by their content
+ assertWrapperRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, field.name());
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ continue;
+ }
+ switch (fieldSchema.getType()) {
+ case RECORD:
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, fieldSchema);
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " record values differ");
+ }
+ break;
+ case ARRAY:
+ assertArrayValuesEqual((List<?>) specificValue, (List<?>)
genericValue, fieldSchema, field.name());
+ break;
+ case MAP:
+ assertMapValuesEqual((Map<?, ?>) specificValue, (Map<?, ?>)
genericValue, fieldSchema, field.name());
+ break;
+ case ENUM:
+ assertEquals(specificValue.toString(), genericValue.toString(),
+ "Field " + field.name() + " enum values differ");
+ break;
+ case BYTES:
+ assertByteArrayEqual((ByteBuffer) specificValue, (ByteBuffer)
genericValue, field.name());
+ break;
+ default:
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ }
+ }
+
+ private static void assertArrayValuesEqual(List<?> specificArray, List<?>
genericArray, Schema schema, String fieldName) {
+ assertEquals(specificArray.size(), genericArray.size(),
+ "Field " + fieldName + " array sizes differ");
+
+ Schema elementSchema = schema.getElementType();
+ for (int i = 0; i < specificArray.size(); i++) {
+ Object specificElement = specificArray.get(i);
+ Object genericElement = genericArray.get(i);
+
+ // Normalize null values
+ if (isNullValue(specificElement)) {
+ specificElement = null;
+ }
+ if (isNullValue(genericElement)) {
+ genericElement = null;
+ }
+ if (specificElement == null && genericElement == null) {
+ continue;
+ }
+ if (specificElement instanceof SpecificRecord && genericElement
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificElement, (GenericRecord)
genericElement, elementSchema);
+ } else if (specificElement instanceof List && genericElement
instanceof List) {
+ // Handle nested lists
+ assertListValuesEqual((List<?>) specificElement, (List<?>)
genericElement, elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificElement, genericElement)) {
+ // Try to compare by string representation as a fallback
+ if (specificElement != null && genericElement != null
+ &&
specificElement.toString().equals(genericElement.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificElement, genericElement,
+ "Field " + fieldName + " array element " + i + " differs");
+ }
+ }
+ }
+ }
+
+ private static void assertMapValuesEqual(Map<?, ?> specificMap, Map<?, ?>
genericMap, Schema schema, String fieldName) {
+ assertEquals(specificMap.size(), genericMap.size(),
+ "Field " + fieldName + " map sizes differ");
+
+ Schema valueSchema = schema.getValueType();
+ for (Map.Entry<?, ?> entry : specificMap.entrySet()) {
+ Object key = entry.getKey();
+ Object specificValue = entry.getValue();
+ Object genericValue = genericMap.get(key);
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue instanceof SpecificRecord && genericValue instanceof
GenericRecord) {
+ // Handle union types by getting the actual record schema
+ Schema recordSchema = valueSchema.getType() == Schema.Type.UNION ?
resolveNullableSchema(valueSchema) : valueSchema;
+ assertRecordsEqual((SpecificRecord) specificValue, (GenericRecord)
genericValue, recordSchema);
+ } else if (specificValue instanceof List && genericValue instanceof
List) {
+ // Handle List comparison by content
+ assertListValuesEqual((List<?>) specificValue, (List<?>)
genericValue, valueSchema, fieldName + "[" + key + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificValue, genericValue)) {
+ // Try to compare by string representation as a fallback
+ if (specificValue != null && genericValue != null
+ && specificValue.toString().equals(genericValue.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + " map value for key " + key + "
differs");
+ }
+ }
+ }
+ }
+
+ private static void assertByteArrayEqual(ByteBuffer specificBytes,
ByteBuffer genericBytes, String fieldName) {
+ if (specificBytes == null && genericBytes == null) {
+ return;
+ }
+ if (specificBytes == null || genericBytes == null) {
+ assertEquals(specificBytes, genericBytes, "Field " + fieldName + "
byte arrays differ");
+ return;
+ }
+ byte[] specificArray = specificBytes.array();
+ byte[] genericArray = genericBytes.array();
+ assertArrayEquals(specificArray, genericArray, "Field " + fieldName + "
byte arrays differ");
+ }
+
+ /**
+ * Check if a value represents null (including JsonProperties$Null)
+ */
+ private static boolean isNullValue(Object value) {
+ return value == null || value instanceof
org.apache.avro.JsonProperties.Null;
+ }
+
+ /**
+ * Compare wrapper records (like BooleanWrapper vs GenericRecord) by their
content
+ */
+ private static void assertWrapperRecordsEqual(SpecificRecord
specificRecord, GenericRecord genericRecord, String fieldName) {
+ // Get the schema from the specific record
+ Schema schema = specificRecord.getSchema();
+
+ // Compare each field in the wrapper
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + "." + field.name() + " values differ");
+ }
+ }
+
+ /**
+ * Compare two lists by content (for cases where they might be different
instances)
+ */
+ private static void assertListValuesEqual(List<?> list1, List<?> list2,
Schema elementSchema, String fieldName) {
+ assertEquals(list1.size(), list2.size(),
+ "Field " + fieldName + " list sizes differ");
+
+ for (int i = 0; i < list1.size(); i++) {
+ Object element1 = list1.get(i);
+ Object element2 = list2.get(i);
+ // Normalize null values
+ if (isNullValue(element1)) {
+ element1 = null;
+ }
+ if (isNullValue(element2)) {
+ element2 = null;
+ }
+ if (element1 == null && element2 == null) {
+ continue;
+ }
+ if (element1 instanceof SpecificRecord && element2 instanceof
GenericRecord) {
+ // For list elements, we need to get the element schema from the
array schema
+ Schema recordSchema = elementSchema.getType() == Schema.Type.ARRAY ?
elementSchema.getElementType() : elementSchema;
+ // Handle union types by getting the actual record schema
+ if (recordSchema.getType() == Schema.Type.UNION) {
+ recordSchema = resolveNullableSchema(recordSchema);
+ }
+ assertRecordsEqual((SpecificRecord) element1, (GenericRecord)
element2, recordSchema);
+ } else if (element1 instanceof List && element2 instanceof List) {
+ assertListValuesEqual((List<?>) element1, (List<?>) element2,
elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(element1, element2)) {
+ // Try to compare by string representation as a fallback
+ if (element1 != null && element2 != null
+ && element1.toString().equals(element2.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(element1, element2,
+ "Field " + fieldName + " list element " + i + " differs");
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Test convertToSpecificRecord for all specified Avro model classes
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordForAllTypes(Class<? extends SpecificRecord>
recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate random generic record
+ GenericRecord genericRecord = AvroTestUtils.generateRandomRecord(schema);
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+
+ /**
+ * Test convertToSpecificRecord with multiple random records for each type
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordMultipleRecords(Class<? extends
SpecificRecord> recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate multiple random records
+ List<GenericRecord> genericRecords =
AvroTestUtils.generateRandomRecords(schema, 3);
+ for (GenericRecord genericRecord : genericRecords) {
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+ }
+
+ /**
+ * Provide all the Avro model classes to test
+ */
+ static Stream<Arguments> provideAvroModelClasses() {
+ return Stream.of(
+
Arguments.of(org.apache.hudi.avro.model.HoodieRollbackPartitionMetadata.class),
Review Comment:
yes, we can.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
Review Comment:
removed.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
Review Comment:
I simplified the code a bit.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
Review Comment:
Interestingly, I did not find any existing functions in hudi.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
Review Comment:
removed.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
Review Comment:
RANDOM.nextDouble(), between 0 ~ 1. Wanted to have some numbers between 0 ~
1000.0
Similarly for RANDOM.nextFloat(), and other integers, doubles.
Just want to make the numbers are within a range.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
+ case STRING:
+ return "test_string_" + RANDOM.nextInt(1000);
+ case BYTES:
+ byte[] bytes = new byte[RANDOM.nextInt(10) + 1];
+ RANDOM.nextBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ case RECORD:
+ return generateRandomRecord(actualSchema);
+ case ENUM:
+ List<String> symbols = actualSchema.getEnumSymbols();
+ return new GenericData.EnumSymbol(actualSchema,
symbols.get(RANDOM.nextInt(symbols.size())));
+ case ARRAY:
+ List<Object> array = new ArrayList<>();
+ int arraySize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < arraySize; i++) {
+ array.add(generateRandomValue(actualSchema.getElementType(),
null));
+ }
+ return array;
+ case MAP:
+ Map<String, Object> map = new HashMap<>();
+ int mapSize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < mapSize; i++) {
+ map.put("key_" + i,
generateRandomValue(actualSchema.getValueType(), null));
+ }
+ return map;
+ case FIXED:
+ byte[] fixedBytes = new byte[actualSchema.getFixedSize()];
+ RANDOM.nextBytes(fixedBytes);
+ return new GenericData.Fixed(actualSchema, fixedBytes);
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Compare a SpecificRecord with a GenericRecord using the same schema
+ */
+ public static void assertRecordsEqual(SpecificRecord specificRecord,
GenericRecord genericRecord, Schema schema) {
+ assertEquals(specificRecord.getClass().getSimpleName(),
genericRecord.getSchema().getName());
+
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values - treat JsonProperties$Null as null
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue == null || genericValue == null) {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " has different null values");
+ continue;
+ }
+ // Handle different types of comparisons based on field schema
+ Schema fieldSchema;
+ try {
+ fieldSchema = resolveNullableSchema(field.schema());
+ } catch (Exception e) {
+ // For complex union types that can't be resolved, do content-based
comparison
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ // Compare wrapper types by their content
+ assertWrapperRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, field.name());
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ continue;
+ }
+ switch (fieldSchema.getType()) {
+ case RECORD:
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, fieldSchema);
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " record values differ");
+ }
+ break;
+ case ARRAY:
+ assertArrayValuesEqual((List<?>) specificValue, (List<?>)
genericValue, fieldSchema, field.name());
+ break;
+ case MAP:
+ assertMapValuesEqual((Map<?, ?>) specificValue, (Map<?, ?>)
genericValue, fieldSchema, field.name());
+ break;
+ case ENUM:
+ assertEquals(specificValue.toString(), genericValue.toString(),
+ "Field " + field.name() + " enum values differ");
+ break;
+ case BYTES:
+ assertByteArrayEqual((ByteBuffer) specificValue, (ByteBuffer)
genericValue, field.name());
+ break;
+ default:
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ }
+ }
+
+ private static void assertArrayValuesEqual(List<?> specificArray, List<?>
genericArray, Schema schema, String fieldName) {
+ assertEquals(specificArray.size(), genericArray.size(),
+ "Field " + fieldName + " array sizes differ");
+
+ Schema elementSchema = schema.getElementType();
+ for (int i = 0; i < specificArray.size(); i++) {
+ Object specificElement = specificArray.get(i);
+ Object genericElement = genericArray.get(i);
+
+ // Normalize null values
+ if (isNullValue(specificElement)) {
+ specificElement = null;
+ }
+ if (isNullValue(genericElement)) {
+ genericElement = null;
+ }
+ if (specificElement == null && genericElement == null) {
+ continue;
+ }
+ if (specificElement instanceof SpecificRecord && genericElement
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificElement, (GenericRecord)
genericElement, elementSchema);
+ } else if (specificElement instanceof List && genericElement
instanceof List) {
+ // Handle nested lists
+ assertListValuesEqual((List<?>) specificElement, (List<?>)
genericElement, elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificElement, genericElement)) {
+ // Try to compare by string representation as a fallback
+ if (specificElement != null && genericElement != null
+ &&
specificElement.toString().equals(genericElement.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificElement, genericElement,
+ "Field " + fieldName + " array element " + i + " differs");
+ }
+ }
+ }
+ }
+
+ private static void assertMapValuesEqual(Map<?, ?> specificMap, Map<?, ?>
genericMap, Schema schema, String fieldName) {
+ assertEquals(specificMap.size(), genericMap.size(),
+ "Field " + fieldName + " map sizes differ");
+
+ Schema valueSchema = schema.getValueType();
+ for (Map.Entry<?, ?> entry : specificMap.entrySet()) {
+ Object key = entry.getKey();
+ Object specificValue = entry.getValue();
+ Object genericValue = genericMap.get(key);
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue instanceof SpecificRecord && genericValue instanceof
GenericRecord) {
+ // Handle union types by getting the actual record schema
+ Schema recordSchema = valueSchema.getType() == Schema.Type.UNION ?
resolveNullableSchema(valueSchema) : valueSchema;
+ assertRecordsEqual((SpecificRecord) specificValue, (GenericRecord)
genericValue, recordSchema);
+ } else if (specificValue instanceof List && genericValue instanceof
List) {
+ // Handle List comparison by content
+ assertListValuesEqual((List<?>) specificValue, (List<?>)
genericValue, valueSchema, fieldName + "[" + key + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificValue, genericValue)) {
+ // Try to compare by string representation as a fallback
+ if (specificValue != null && genericValue != null
+ && specificValue.toString().equals(genericValue.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + " map value for key " + key + "
differs");
+ }
+ }
+ }
+ }
+
+ private static void assertByteArrayEqual(ByteBuffer specificBytes,
ByteBuffer genericBytes, String fieldName) {
+ if (specificBytes == null && genericBytes == null) {
+ return;
+ }
+ if (specificBytes == null || genericBytes == null) {
+ assertEquals(specificBytes, genericBytes, "Field " + fieldName + "
byte arrays differ");
+ return;
+ }
+ byte[] specificArray = specificBytes.array();
+ byte[] genericArray = genericBytes.array();
+ assertArrayEquals(specificArray, genericArray, "Field " + fieldName + "
byte arrays differ");
+ }
+
+ /**
+ * Check if a value represents null (including JsonProperties$Null)
+ */
+ private static boolean isNullValue(Object value) {
+ return value == null || value instanceof
org.apache.avro.JsonProperties.Null;
+ }
+
+ /**
+ * Compare wrapper records (like BooleanWrapper vs GenericRecord) by their
content
+ */
+ private static void assertWrapperRecordsEqual(SpecificRecord
specificRecord, GenericRecord genericRecord, String fieldName) {
+ // Get the schema from the specific record
+ Schema schema = specificRecord.getSchema();
+
+ // Compare each field in the wrapper
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + "." + field.name() + " values differ");
+ }
+ }
+
+ /**
+ * Compare two lists by content (for cases where they might be different
instances)
+ */
+ private static void assertListValuesEqual(List<?> list1, List<?> list2,
Schema elementSchema, String fieldName) {
+ assertEquals(list1.size(), list2.size(),
+ "Field " + fieldName + " list sizes differ");
+
+ for (int i = 0; i < list1.size(); i++) {
+ Object element1 = list1.get(i);
+ Object element2 = list2.get(i);
+ // Normalize null values
+ if (isNullValue(element1)) {
+ element1 = null;
+ }
+ if (isNullValue(element2)) {
+ element2 = null;
+ }
+ if (element1 == null && element2 == null) {
+ continue;
+ }
+ if (element1 instanceof SpecificRecord && element2 instanceof
GenericRecord) {
+ // For list elements, we need to get the element schema from the
array schema
+ Schema recordSchema = elementSchema.getType() == Schema.Type.ARRAY ?
elementSchema.getElementType() : elementSchema;
+ // Handle union types by getting the actual record schema
+ if (recordSchema.getType() == Schema.Type.UNION) {
+ recordSchema = resolveNullableSchema(recordSchema);
+ }
+ assertRecordsEqual((SpecificRecord) element1, (GenericRecord)
element2, recordSchema);
+ } else if (element1 instanceof List && element2 instanceof List) {
+ assertListValuesEqual((List<?>) element1, (List<?>) element2,
elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(element1, element2)) {
+ // Try to compare by string representation as a fallback
+ if (element1 != null && element2 != null
+ && element1.toString().equals(element2.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(element1, element2,
+ "Field " + fieldName + " list element " + i + " differs");
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Test convertToSpecificRecord for all specified Avro model classes
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordForAllTypes(Class<? extends SpecificRecord>
recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate random generic record
+ GenericRecord genericRecord = AvroTestUtils.generateRandomRecord(schema);
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+
+ /**
+ * Test convertToSpecificRecord with multiple random records for each type
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordMultipleRecords(Class<? extends
SpecificRecord> recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate multiple random records
+ List<GenericRecord> genericRecords =
AvroTestUtils.generateRandomRecords(schema, 3);
+ for (GenericRecord genericRecord : genericRecords) {
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
Review Comment:
yes
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
+ case STRING:
+ return "test_string_" + RANDOM.nextInt(1000);
+ case BYTES:
+ byte[] bytes = new byte[RANDOM.nextInt(10) + 1];
+ RANDOM.nextBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ case RECORD:
+ return generateRandomRecord(actualSchema);
+ case ENUM:
+ List<String> symbols = actualSchema.getEnumSymbols();
+ return new GenericData.EnumSymbol(actualSchema,
symbols.get(RANDOM.nextInt(symbols.size())));
+ case ARRAY:
+ List<Object> array = new ArrayList<>();
+ int arraySize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < arraySize; i++) {
+ array.add(generateRandomValue(actualSchema.getElementType(),
null));
+ }
+ return array;
+ case MAP:
+ Map<String, Object> map = new HashMap<>();
+ int mapSize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < mapSize; i++) {
+ map.put("key_" + i,
generateRandomValue(actualSchema.getValueType(), null));
+ }
+ return map;
+ case FIXED:
+ byte[] fixedBytes = new byte[actualSchema.getFixedSize()];
+ RANDOM.nextBytes(fixedBytes);
+ return new GenericData.Fixed(actualSchema, fixedBytes);
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Compare a SpecificRecord with a GenericRecord using the same schema
+ */
+ public static void assertRecordsEqual(SpecificRecord specificRecord,
GenericRecord genericRecord, Schema schema) {
+ assertEquals(specificRecord.getClass().getSimpleName(),
genericRecord.getSchema().getName());
+
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values - treat JsonProperties$Null as null
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue == null || genericValue == null) {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " has different null values");
+ continue;
+ }
+ // Handle different types of comparisons based on field schema
+ Schema fieldSchema;
+ try {
+ fieldSchema = resolveNullableSchema(field.schema());
+ } catch (Exception e) {
+ // For complex union types that can't be resolved, do content-based
comparison
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ // Compare wrapper types by their content
+ assertWrapperRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, field.name());
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ continue;
+ }
+ switch (fieldSchema.getType()) {
+ case RECORD:
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, fieldSchema);
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " record values differ");
+ }
+ break;
+ case ARRAY:
+ assertArrayValuesEqual((List<?>) specificValue, (List<?>)
genericValue, fieldSchema, field.name());
+ break;
+ case MAP:
+ assertMapValuesEqual((Map<?, ?>) specificValue, (Map<?, ?>)
genericValue, fieldSchema, field.name());
+ break;
+ case ENUM:
+ assertEquals(specificValue.toString(), genericValue.toString(),
+ "Field " + field.name() + " enum values differ");
+ break;
+ case BYTES:
+ assertByteArrayEqual((ByteBuffer) specificValue, (ByteBuffer)
genericValue, field.name());
+ break;
+ default:
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ }
+ }
+
+ private static void assertArrayValuesEqual(List<?> specificArray, List<?>
genericArray, Schema schema, String fieldName) {
+ assertEquals(specificArray.size(), genericArray.size(),
+ "Field " + fieldName + " array sizes differ");
+
+ Schema elementSchema = schema.getElementType();
+ for (int i = 0; i < specificArray.size(); i++) {
+ Object specificElement = specificArray.get(i);
+ Object genericElement = genericArray.get(i);
+
+ // Normalize null values
+ if (isNullValue(specificElement)) {
+ specificElement = null;
+ }
+ if (isNullValue(genericElement)) {
+ genericElement = null;
+ }
+ if (specificElement == null && genericElement == null) {
+ continue;
+ }
+ if (specificElement instanceof SpecificRecord && genericElement
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificElement, (GenericRecord)
genericElement, elementSchema);
+ } else if (specificElement instanceof List && genericElement
instanceof List) {
+ // Handle nested lists
+ assertListValuesEqual((List<?>) specificElement, (List<?>)
genericElement, elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificElement, genericElement)) {
+ // Try to compare by string representation as a fallback
+ if (specificElement != null && genericElement != null
+ &&
specificElement.toString().equals(genericElement.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificElement, genericElement,
+ "Field " + fieldName + " array element " + i + " differs");
+ }
+ }
+ }
+ }
+
+ private static void assertMapValuesEqual(Map<?, ?> specificMap, Map<?, ?>
genericMap, Schema schema, String fieldName) {
+ assertEquals(specificMap.size(), genericMap.size(),
+ "Field " + fieldName + " map sizes differ");
+
+ Schema valueSchema = schema.getValueType();
+ for (Map.Entry<?, ?> entry : specificMap.entrySet()) {
+ Object key = entry.getKey();
+ Object specificValue = entry.getValue();
+ Object genericValue = genericMap.get(key);
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue instanceof SpecificRecord && genericValue instanceof
GenericRecord) {
+ // Handle union types by getting the actual record schema
+ Schema recordSchema = valueSchema.getType() == Schema.Type.UNION ?
resolveNullableSchema(valueSchema) : valueSchema;
+ assertRecordsEqual((SpecificRecord) specificValue, (GenericRecord)
genericValue, recordSchema);
+ } else if (specificValue instanceof List && genericValue instanceof
List) {
+ // Handle List comparison by content
+ assertListValuesEqual((List<?>) specificValue, (List<?>)
genericValue, valueSchema, fieldName + "[" + key + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificValue, genericValue)) {
+ // Try to compare by string representation as a fallback
+ if (specificValue != null && genericValue != null
+ && specificValue.toString().equals(genericValue.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + " map value for key " + key + "
differs");
+ }
+ }
+ }
+ }
+
+ private static void assertByteArrayEqual(ByteBuffer specificBytes,
ByteBuffer genericBytes, String fieldName) {
+ if (specificBytes == null && genericBytes == null) {
+ return;
+ }
+ if (specificBytes == null || genericBytes == null) {
+ assertEquals(specificBytes, genericBytes, "Field " + fieldName + "
byte arrays differ");
+ return;
+ }
+ byte[] specificArray = specificBytes.array();
+ byte[] genericArray = genericBytes.array();
+ assertArrayEquals(specificArray, genericArray, "Field " + fieldName + "
byte arrays differ");
+ }
+
+ /**
+ * Check if a value represents null (including JsonProperties$Null)
+ */
+ private static boolean isNullValue(Object value) {
+ return value == null || value instanceof
org.apache.avro.JsonProperties.Null;
+ }
+
+ /**
+ * Compare wrapper records (like BooleanWrapper vs GenericRecord) by their
content
+ */
+ private static void assertWrapperRecordsEqual(SpecificRecord
specificRecord, GenericRecord genericRecord, String fieldName) {
+ // Get the schema from the specific record
+ Schema schema = specificRecord.getSchema();
+
+ // Compare each field in the wrapper
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + "." + field.name() + " values differ");
+ }
+ }
+
+ /**
+ * Compare two lists by content (for cases where they might be different
instances)
+ */
+ private static void assertListValuesEqual(List<?> list1, List<?> list2,
Schema elementSchema, String fieldName) {
+ assertEquals(list1.size(), list2.size(),
+ "Field " + fieldName + " list sizes differ");
+
+ for (int i = 0; i < list1.size(); i++) {
+ Object element1 = list1.get(i);
+ Object element2 = list2.get(i);
+ // Normalize null values
+ if (isNullValue(element1)) {
+ element1 = null;
+ }
+ if (isNullValue(element2)) {
+ element2 = null;
+ }
+ if (element1 == null && element2 == null) {
+ continue;
+ }
+ if (element1 instanceof SpecificRecord && element2 instanceof
GenericRecord) {
+ // For list elements, we need to get the element schema from the
array schema
+ Schema recordSchema = elementSchema.getType() == Schema.Type.ARRAY ?
elementSchema.getElementType() : elementSchema;
+ // Handle union types by getting the actual record schema
+ if (recordSchema.getType() == Schema.Type.UNION) {
+ recordSchema = resolveNullableSchema(recordSchema);
+ }
+ assertRecordsEqual((SpecificRecord) element1, (GenericRecord)
element2, recordSchema);
+ } else if (element1 instanceof List && element2 instanceof List) {
+ assertListValuesEqual((List<?>) element1, (List<?>) element2,
elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(element1, element2)) {
+ // Try to compare by string representation as a fallback
+ if (element1 != null && element2 != null
+ && element1.toString().equals(element2.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(element1, element2,
+ "Field " + fieldName + " list element " + i + " differs");
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Test convertToSpecificRecord for all specified Avro model classes
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordForAllTypes(Class<? extends SpecificRecord>
recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate random generic record
+ GenericRecord genericRecord = AvroTestUtils.generateRandomRecord(schema);
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+
+ /**
+ * Test convertToSpecificRecord with multiple random records for each type
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordMultipleRecords(Class<? extends
SpecificRecord> recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate multiple random records
+ List<GenericRecord> genericRecords =
AvroTestUtils.generateRandomRecords(schema, 3);
+ for (GenericRecord genericRecord : genericRecords) {
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
Review Comment:
Done.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
+ case STRING:
+ return "test_string_" + RANDOM.nextInt(1000);
+ case BYTES:
+ byte[] bytes = new byte[RANDOM.nextInt(10) + 1];
+ RANDOM.nextBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ case RECORD:
+ return generateRandomRecord(actualSchema);
+ case ENUM:
+ List<String> symbols = actualSchema.getEnumSymbols();
+ return new GenericData.EnumSymbol(actualSchema,
symbols.get(RANDOM.nextInt(symbols.size())));
+ case ARRAY:
+ List<Object> array = new ArrayList<>();
+ int arraySize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < arraySize; i++) {
+ array.add(generateRandomValue(actualSchema.getElementType(),
null));
+ }
+ return array;
+ case MAP:
+ Map<String, Object> map = new HashMap<>();
+ int mapSize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < mapSize; i++) {
+ map.put("key_" + i,
generateRandomValue(actualSchema.getValueType(), null));
+ }
+ return map;
+ case FIXED:
+ byte[] fixedBytes = new byte[actualSchema.getFixedSize()];
+ RANDOM.nextBytes(fixedBytes);
+ return new GenericData.Fixed(actualSchema, fixedBytes);
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Compare a SpecificRecord with a GenericRecord using the same schema
+ */
+ public static void assertRecordsEqual(SpecificRecord specificRecord,
GenericRecord genericRecord, Schema schema) {
+ assertEquals(specificRecord.getClass().getSimpleName(),
genericRecord.getSchema().getName());
+
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values - treat JsonProperties$Null as null
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue == null || genericValue == null) {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " has different null values");
+ continue;
+ }
+ // Handle different types of comparisons based on field schema
+ Schema fieldSchema;
+ try {
+ fieldSchema = resolveNullableSchema(field.schema());
+ } catch (Exception e) {
+ // For complex union types that can't be resolved, do content-based
comparison
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ // Compare wrapper types by their content
+ assertWrapperRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, field.name());
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ continue;
+ }
+ switch (fieldSchema.getType()) {
+ case RECORD:
+ if (specificValue instanceof SpecificRecord && genericValue
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificValue,
(GenericRecord) genericValue, fieldSchema);
+ } else {
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " record values differ");
+ }
+ break;
+ case ARRAY:
+ assertArrayValuesEqual((List<?>) specificValue, (List<?>)
genericValue, fieldSchema, field.name());
+ break;
+ case MAP:
+ assertMapValuesEqual((Map<?, ?>) specificValue, (Map<?, ?>)
genericValue, fieldSchema, field.name());
+ break;
+ case ENUM:
+ assertEquals(specificValue.toString(), genericValue.toString(),
+ "Field " + field.name() + " enum values differ");
+ break;
+ case BYTES:
+ assertByteArrayEqual((ByteBuffer) specificValue, (ByteBuffer)
genericValue, field.name());
+ break;
+ default:
+ assertEquals(specificValue, genericValue,
+ "Field " + field.name() + " values differ");
+ }
+ }
+ }
+
+ private static void assertArrayValuesEqual(List<?> specificArray, List<?>
genericArray, Schema schema, String fieldName) {
+ assertEquals(specificArray.size(), genericArray.size(),
+ "Field " + fieldName + " array sizes differ");
+
+ Schema elementSchema = schema.getElementType();
+ for (int i = 0; i < specificArray.size(); i++) {
+ Object specificElement = specificArray.get(i);
+ Object genericElement = genericArray.get(i);
+
+ // Normalize null values
+ if (isNullValue(specificElement)) {
+ specificElement = null;
+ }
+ if (isNullValue(genericElement)) {
+ genericElement = null;
+ }
+ if (specificElement == null && genericElement == null) {
+ continue;
+ }
+ if (specificElement instanceof SpecificRecord && genericElement
instanceof GenericRecord) {
+ assertRecordsEqual((SpecificRecord) specificElement, (GenericRecord)
genericElement, elementSchema);
+ } else if (specificElement instanceof List && genericElement
instanceof List) {
+ // Handle nested lists
+ assertListValuesEqual((List<?>) specificElement, (List<?>)
genericElement, elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificElement, genericElement)) {
+ // Try to compare by string representation as a fallback
+ if (specificElement != null && genericElement != null
+ &&
specificElement.toString().equals(genericElement.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificElement, genericElement,
+ "Field " + fieldName + " array element " + i + " differs");
+ }
+ }
+ }
+ }
+
+ private static void assertMapValuesEqual(Map<?, ?> specificMap, Map<?, ?>
genericMap, Schema schema, String fieldName) {
+ assertEquals(specificMap.size(), genericMap.size(),
+ "Field " + fieldName + " map sizes differ");
+
+ Schema valueSchema = schema.getValueType();
+ for (Map.Entry<?, ?> entry : specificMap.entrySet()) {
+ Object key = entry.getKey();
+ Object specificValue = entry.getValue();
+ Object genericValue = genericMap.get(key);
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ if (specificValue == null && genericValue == null) {
+ continue;
+ }
+ if (specificValue instanceof SpecificRecord && genericValue instanceof
GenericRecord) {
+ // Handle union types by getting the actual record schema
+ Schema recordSchema = valueSchema.getType() == Schema.Type.UNION ?
resolveNullableSchema(valueSchema) : valueSchema;
+ assertRecordsEqual((SpecificRecord) specificValue, (GenericRecord)
genericValue, recordSchema);
+ } else if (specificValue instanceof List && genericValue instanceof
List) {
+ // Handle List comparison by content
+ assertListValuesEqual((List<?>) specificValue, (List<?>)
genericValue, valueSchema, fieldName + "[" + key + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(specificValue, genericValue)) {
+ // Try to compare by string representation as a fallback
+ if (specificValue != null && genericValue != null
+ && specificValue.toString().equals(genericValue.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + " map value for key " + key + "
differs");
+ }
+ }
+ }
+ }
+
+ private static void assertByteArrayEqual(ByteBuffer specificBytes,
ByteBuffer genericBytes, String fieldName) {
+ if (specificBytes == null && genericBytes == null) {
+ return;
+ }
+ if (specificBytes == null || genericBytes == null) {
+ assertEquals(specificBytes, genericBytes, "Field " + fieldName + "
byte arrays differ");
+ return;
+ }
+ byte[] specificArray = specificBytes.array();
+ byte[] genericArray = genericBytes.array();
+ assertArrayEquals(specificArray, genericArray, "Field " + fieldName + "
byte arrays differ");
+ }
+
+ /**
+ * Check if a value represents null (including JsonProperties$Null)
+ */
+ private static boolean isNullValue(Object value) {
+ return value == null || value instanceof
org.apache.avro.JsonProperties.Null;
+ }
+
+ /**
+ * Compare wrapper records (like BooleanWrapper vs GenericRecord) by their
content
+ */
+ private static void assertWrapperRecordsEqual(SpecificRecord
specificRecord, GenericRecord genericRecord, String fieldName) {
+ // Get the schema from the specific record
+ Schema schema = specificRecord.getSchema();
+
+ // Compare each field in the wrapper
+ for (Schema.Field field : schema.getFields()) {
+ Object specificValue = specificRecord.get(field.pos());
+ Object genericValue = genericRecord.get(field.pos());
+
+ // Normalize null values
+ if (isNullValue(specificValue)) {
+ specificValue = null;
+ }
+ if (isNullValue(genericValue)) {
+ genericValue = null;
+ }
+ assertEquals(specificValue, genericValue,
+ "Field " + fieldName + "." + field.name() + " values differ");
+ }
+ }
+
+ /**
+ * Compare two lists by content (for cases where they might be different
instances)
+ */
+ private static void assertListValuesEqual(List<?> list1, List<?> list2,
Schema elementSchema, String fieldName) {
+ assertEquals(list1.size(), list2.size(),
+ "Field " + fieldName + " list sizes differ");
+
+ for (int i = 0; i < list1.size(); i++) {
+ Object element1 = list1.get(i);
+ Object element2 = list2.get(i);
+ // Normalize null values
+ if (isNullValue(element1)) {
+ element1 = null;
+ }
+ if (isNullValue(element2)) {
+ element2 = null;
+ }
+ if (element1 == null && element2 == null) {
+ continue;
+ }
+ if (element1 instanceof SpecificRecord && element2 instanceof
GenericRecord) {
+ // For list elements, we need to get the element schema from the
array schema
+ Schema recordSchema = elementSchema.getType() == Schema.Type.ARRAY ?
elementSchema.getElementType() : elementSchema;
+ // Handle union types by getting the actual record schema
+ if (recordSchema.getType() == Schema.Type.UNION) {
+ recordSchema = resolveNullableSchema(recordSchema);
+ }
+ assertRecordsEqual((SpecificRecord) element1, (GenericRecord)
element2, recordSchema);
+ } else if (element1 instanceof List && element2 instanceof List) {
+ assertListValuesEqual((List<?>) element1, (List<?>) element2,
elementSchema, fieldName + "[" + i + "]");
+ } else {
+ // For other types, compare by content if they're different
instances but same content
+ if (!Objects.equals(element1, element2)) {
+ // Try to compare by string representation as a fallback
+ if (element1 != null && element2 != null
+ && element1.toString().equals(element2.toString())) {
+ // They have the same content, so they're equivalent
+ continue;
+ }
+ assertEquals(element1, element2,
+ "Field " + fieldName + " list element " + i + " differs");
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Test convertToSpecificRecord for all specified Avro model classes
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordForAllTypes(Class<? extends SpecificRecord>
recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate random generic record
+ GenericRecord genericRecord = AvroTestUtils.generateRandomRecord(schema);
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+
+ /**
+ * Test convertToSpecificRecord with multiple random records for each type
+ */
+ @ParameterizedTest
+ @MethodSource("provideAvroModelClasses")
+ void testConvertToSpecificRecordMultipleRecords(Class<? extends
SpecificRecord> recordClass) {
+ // Get schema from the class
+ Schema schema = SpecificData.get().getSchema(recordClass);
+ // Generate multiple random records
+ List<GenericRecord> genericRecords =
AvroTestUtils.generateRandomRecords(schema, 3);
+ for (GenericRecord genericRecord : genericRecords) {
+ // Convert to specific record - cast to the expected type
+ @SuppressWarnings("unchecked")
+ Class<? extends org.apache.avro.specific.SpecificRecordBase>
specificRecordBaseClass =
+ (Class<? extends org.apache.avro.specific.SpecificRecordBase>)
recordClass;
+ SpecificRecord specificRecord =
HoodieAvroUtils.convertToSpecificRecord(specificRecordBaseClass, genericRecord);
+ // Assert the output class
+ assertEquals(recordClass, specificRecord.getClass());
+ // Compare the records
+ AvroTestUtils.assertRecordsEqual(specificRecord, genericRecord, schema);
+ }
+ }
+
+ /**
+ * Provide all the Avro model classes to test
+ */
+ static Stream<Arguments> provideAvroModelClasses() {
+ return Stream.of(
+
Arguments.of(org.apache.hudi.avro.model.HoodieRollbackPartitionMetadata.class),
Review Comment:
done.
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java:
##########
@@ -1231,4 +1234,458 @@ void
recordNeedsRewriteForExtendedAvroTypePromotion(Schema writerSchema, Schema
boolean result =
HoodieAvroUtils.recordNeedsRewriteForExtendedAvroTypePromotion(writerSchema,
readerSchema);
assertEquals(expected, result);
}
+
+ /**
+ * Utility class for generating random GenericRecord instances and comparing
records
+ */
+ private static class AvroTestUtils {
+ private static final Random RANDOM = new Random(42); // Fixed seed for
reproducible tests
+
+ /**
+ * Generate a random GenericRecord for the given schema
+ */
+ public static GenericRecord generateRandomRecord(Schema schema) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (Schema.Field field : schema.getFields()) {
+ Object value = generateRandomValue(field.schema(), field.defaultVal());
+ record.put(field.pos(), value);
+ }
+ return record;
+ }
+
+ /**
+ * Generate a list of random GenericRecord instances
+ */
+ public static List<GenericRecord> generateRandomRecords(Schema schema, int
count) {
+ List<GenericRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(generateRandomRecord(schema));
+ }
+ return records;
+ }
+
+ /**
+ * Generate a random value for the given schema type
+ */
+ private static Object generateRandomValue(Schema schema, Object
defaultValue) {
+ // Handle union types first
+ if (schema.getType() == Schema.Type.UNION) {
+ List<Schema> types = schema.getTypes();
+ // For nullable unions, sometimes return null
+ if (types.size() == 2 && types.get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ // For complex unions (more than 2 types), never return null to avoid
HoodieAvroSchemaException
+ if (types.size() > 2) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // For 2-type unions that are not simple nullable unions, also avoid
null
+ if (types.size() == 2 && !(types.get(0).getType() == Schema.Type.NULL
|| types.get(1).getType() == Schema.Type.NULL)) {
+ // Choose the first non-null type
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Choose a non-null type from the union
+ Schema nonNullType = types.stream()
+ .filter(t -> t.getType() != Schema.Type.NULL)
+ .findFirst()
+ .orElse(types.get(0));
+ return generateRandomValue(nonNullType, null);
+ }
+ // Handle default values
+ if (defaultValue != null
+ && !(defaultValue instanceof org.apache.avro.JsonProperties.Null)
+ && RANDOM.nextBoolean()) {
+ return defaultValue;
+ }
+ // For nullable fields, sometimes return null
+ if (schema.getType() == Schema.Type.UNION && schema.getTypes().size() ==
2
+ && schema.getTypes().get(0).getType() == Schema.Type.NULL &&
RANDOM.nextBoolean()) {
+ return null;
+ }
+ Schema actualSchema = schema;
+ try {
+ actualSchema = resolveNullableSchema(schema);
+ } catch (Exception e) {
+ // If we can't resolve the schema, just use the original
+ actualSchema = schema;
+ }
+
+ switch (actualSchema.getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return RANDOM.nextBoolean();
+ case INT:
+ return RANDOM.nextInt(1000);
+ case LONG:
+ return RANDOM.nextLong() % 1000000L;
+ case FLOAT:
+ return RANDOM.nextFloat() * 100f;
+ case DOUBLE:
+ return RANDOM.nextDouble() * 1000.0;
+ case STRING:
+ return "test_string_" + RANDOM.nextInt(1000);
+ case BYTES:
+ byte[] bytes = new byte[RANDOM.nextInt(10) + 1];
+ RANDOM.nextBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ case RECORD:
+ return generateRandomRecord(actualSchema);
+ case ENUM:
+ List<String> symbols = actualSchema.getEnumSymbols();
+ return new GenericData.EnumSymbol(actualSchema,
symbols.get(RANDOM.nextInt(symbols.size())));
+ case ARRAY:
+ List<Object> array = new ArrayList<>();
+ int arraySize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < arraySize; i++) {
+ array.add(generateRandomValue(actualSchema.getElementType(),
null));
+ }
+ return array;
+ case MAP:
+ Map<String, Object> map = new HashMap<>();
+ int mapSize = RANDOM.nextInt(3) + 1;
+ for (int i = 0; i < mapSize; i++) {
+ map.put("key_" + i,
generateRandomValue(actualSchema.getValueType(), null));
+ }
+ return map;
+ case FIXED:
+ byte[] fixedBytes = new byte[actualSchema.getFixedSize()];
+ RANDOM.nextBytes(fixedBytes);
+ return new GenericData.Fixed(actualSchema, fixedBytes);
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Compare a SpecificRecord with a GenericRecord using the same schema
+ */
+ public static void assertRecordsEqual(SpecificRecord specificRecord,
GenericRecord genericRecord, Schema schema) {
Review Comment:
Tried, and worked. How do you know?!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
