Revert to generate avro schema to record of value + metadata and pass in variant value + metadata schema

Author: sfc-gh-aixu

parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java

@@ -24,6 +24,7 @@
 import org.apache.avro.Conversion;
 import org.apache.avro.LogicalType;
 import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
 import org.apache.avro.generic.GenericArray;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.IndexedRecord;
@@ -58,6 +59,18 @@ class AvroIndexedRecordConverter<T extends IndexedRecord> extends GroupConverter
   private final GenericData model;
   private final Map<Schema.Field, Object> recordDefaults = new HashMap<Schema.Field, Object>();
 
+  private static final Schema VARIANT_SCHEMA = SchemaBuilder.record("VariantRecord")
+      .fields()
+      .name("metadata")
+      .type()
+      .bytesType()
+      .noDefault()
+      .name("value")
+      .type()
+      .bytesType()
+      .noDefault()
+      .endRecord();
+
   public AvroIndexedRecordConverter(MessageType parquetSchema, Schema avroSchema) {
     this(parquetSchema, avroSchema, SpecificData.get());
   }
@@ -170,7 +183,7 @@ private static Converter newConverter(Schema schema, Type type, GenericData mode
         return new MapConverter(parent, type.asGroupType(), schema, model);
       case RECORD:
         if (type.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.VariantLogicalTypeAnnotation) {
-          return new AvroVariantConverter(parent, type.asGroupType(), schema, model);
+          return new AvroVariantConverter(parent, type.asGroupType(), VARIANT_SCHEMA, model);
         } else {
           return new AvroIndexedRecordConverter<>(parent, type.asGroupType(), schema, model);
         }

parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java

@@ -93,6 +93,18 @@ class AvroRecordConverter<T> extends AvroConverters.AvroGroupConverter {
   private final GenericData model;
   private final Map<Schema.Field, Object> recordDefaults = new HashMap<Schema.Field, Object>();
 
+  private static final Schema VARIANT_SCHEMA = SchemaBuilder.record("VariantRecord")
+      .fields()
+      .name("metadata")
+      .type()
+      .bytesType()
+      .noDefault()
+      .name("value")
+      .type()
+      .bytesType()
+      .noDefault()
+      .endRecord();
+
   AvroRecordConverter(
       MessageType parquetSchema, Schema avroSchema, GenericData baseModel, ReflectClassValidator validator) {
     this(null, parquetSchema, avroSchema, baseModel, validator);
@@ -396,7 +408,7 @@ private static Converter newConverter(
         return newStringConverter(schema, model, parent, validator);
       case RECORD:
         if (type.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.VariantLogicalTypeAnnotation) {
-          return new AvroVariantConverter(parent, type.asGroupType(), schema, model);
+          return new AvroVariantConverter(parent, type.asGroupType(), VARIANT_SCHEMA, model);
         } else {
           return new AvroRecordConverter(parent, type.asGroupType(), schema, model, validator);
         }

parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java

@@ -470,8 +470,11 @@ public Optional<Schema> visit(
               @Override
               public Optional<Schema> visit(
                   LogicalTypeAnnotation.VariantLogicalTypeAnnotation variantLogicalType) {
-                return of(
-                    convertFields(parquetGroupType.getName(), parquetGroupType.getFields(), names));
+                String name = parquetGroupType.getName();
+                List<Schema.Field> fields = new ArrayList<>();
+                fields.add(new Schema.Field("metadata", Schema.create(Schema.Type.BYTES)));
+                fields.add(new Schema.Field("value", Schema.create(Schema.Type.BYTES)));
+                return of(Schema.createRecord(name, null, namespace(name, names), false, fields));
               }
             })
             .orElseThrow(

parquet-avro/src/main/java/org/apache/parquet/avro/AvroVariantConverter.java

@@ -21,7 +21,6 @@
 import java.nio.ByteBuffer;
 import java.util.function.Consumer;
 import org.apache.avro.Schema;
-import org.apache.avro.SchemaBuilder;
 import org.apache.avro.generic.GenericData;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.io.api.Converter;
@@ -35,27 +34,21 @@
  * Converter for Variant values.
  */
 class AvroVariantConverter extends GroupConverter implements VariantConverters.ParentConverter<VariantBuilder> {
-  private static final Schema VARIANT_SCHEMA = SchemaBuilder.record("VariantRecord")
-      .fields()
-      .name("metadata")
-      .type()
-      .bytesType()
-      .noDefault()
-      .name("value")
-      .type()
-      .bytesType()
-      .noDefault()
-      .endRecord();
-
   private final ParentValueContainer parent;
+  private final Schema avroSchema;
   private final GenericData model;
+  private final int metadataPos;
+  private final int valuePos;
   private final GroupConverter wrappedConverter;
 
   private VariantBuilder builder = null;
   private ImmutableMetadata metadata = null;
 
   AvroVariantConverter(ParentValueContainer parent, GroupType variantGroup, Schema avroSchema, GenericData model) {
     this.parent = parent;
+    this.avroSchema = avroSchema;
+    this.metadataPos = avroSchema.getField("metadata").pos();
+    this.valuePos = avroSchema.getField("value").pos();
     this.model = model;
     this.wrappedConverter = VariantConverters.newVariantConverter(variantGroup, this::setMetadata, this);
   }
@@ -84,9 +77,9 @@ public void end() {
 
     builder.appendNullIfEmpty();
 
-    Object record = model.newRecord(null, VARIANT_SCHEMA);
-    model.setField(record, "metadata", 0, metadata.getEncodedBuffer());
-    model.setField(record, "value", 1, builder.encodedValue());
+    Object record = model.newRecord(null, avroSchema);
+    model.setField(record, "metadata", metadataPos, metadata.getEncodedBuffer());
+    model.setField(record, "value", valuePos, builder.encodedValue());
     parent.add(record);
 
     this.builder = null;