parquet-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jul...@apache.org
Subject [parquet-mr] branch master updated: PARQUET-968 Add Hive/Presto support in ProtoParquet
Date Thu, 26 Apr 2018 12:48:28 GMT
This is an automated email from the ASF dual-hosted git repository.

julien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-mr.git


The following commit(s) were added to refs/heads/master by this push:
     new f849384  PARQUET-968 Add Hive/Presto support in ProtoParquet
f849384 is described below

commit f84938441be49c665595c936ac631c3e5f171bf9
Author: Constantin Muraru <cmuraru@adobe.com>
AuthorDate: Thu Apr 26 08:48:08 2018 -0400

    PARQUET-968 Add Hive/Presto support in ProtoParquet
    
    This PR adds Hive (https://github.com/apache/hive) and Presto (https://github.com/prestodb/presto) support for parquet messages written with ProtoParquetWriter. Hive and other tools, such as Presto (used by AWS Athena), rely on specific LIST/MAP wrappers (as defined in the parquet spec: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md). These wrappers are currently missing from the ProtoParquet schema. AvroParquet works just fine, because it adds these wrappers whe [...]
    
    This is backward compatible. Messages written without the extra LIST/MAP wrappers are still being read successfully using the updated ProtoParquetReader.
    
    Regarding the change.
    Given the following protobuf schema:
    
    ```
    message ListOfPrimitives {
        repeated int64 my_repeated_id = 1;
    }
    ```
    
    Old parquet schema was:
    ```
    message ListOfPrimitives {
      repeated int64 my_repeated_id = 1;
    }
    ```
    
    New parquet schema is:
    ```
    message ListOfPrimitives {
      required group my_repeated_id (LIST) = 1 {
        repeated group list {
          required int64 element;
        }
      }
    }
    ```
    ---
    
    For list of messages, the changes look like this:
    
    Protobuf schema:
    ```
    message ListOfMessages {
        string top_field = 1;
        repeated MyInnerMessage first_array = 2;
    }
    
    message MyInnerMessage {
        int32 inner_field = 1;
    }
    ```
    
    Old parquet schema was:
    ```
    message TestProto3.ListOfMessages {
      optional binary top_field (UTF8) = 1;
      repeated group first_array = 2 {
        optional int32 inner_field = 1;
      }
    }
    ```
    
    The expected parquet schema, compatible with Hive (and similar to parquet-avro) is the following (notice the LIST wrapper):
    
    ```
    message TestProto3.ListOfMessages {
      optional binary top_field (UTF8) = 1;
      required group first_array (LIST) = 2 {
        repeated group list {
          optional group element {
            optional int32 inner_field = 1;
          }
        }
      }
    }
    ```
    
    ---
    
    Similar for maps. Protobuf schema:
    ```
    message TopMessage {
        map<int64, MyInnerMessage> myMap = 1;
    }
    
    message MyInnerMessage {
        int32 inner_field = 1;
    }
    ```
    
    Old parquet schema:
    ```
    message TestProto3.TopMessage {
      repeated group myMap = 1 {
        optional int64 key = 1;
        optional group value = 2 {
          optional int32 inner_field = 1;
        }
      }
    }
    ```
    
    New parquet schema (notice the `MAP` wrapper):
    ```
    message TestProto3.TopMessage {
      required group myMap (MAP) = 1 {
        repeated group key_value {
          required int64 key;
          optional group value {
            optional int32 inner_field = 1;
          }
        }
      }
    }
    ```
    
    Jira: https://issues.apache.org/jira/browse/PARQUET-968
    
    Author: Constantin Muraru <cmuraru@adobe.com>
    Author: Benoît Hanotte <BenoitHanotte@users.noreply.github.com>
    
    Closes #411 from costimuraru/PARQUET-968 and squashes the following commits:
    
    16eafcb6 [Benoît Hanotte] PARQUET-968 add proto flag to enable writing using specs-compliant schemas (#2)
    a8bd7041 [Constantin Muraru] Pick up commit from @andredasilvapinto
    5cf92487 [Constantin Muraru] PARQUET-968 Add Hive support in ProtoParquet
---
 .../parquet/proto/ProtoMessageConverter.java       | 126 ++++-
 .../apache/parquet/proto/ProtoSchemaConverter.java | 169 ++++--
 .../apache/parquet/proto/ProtoWriteSupport.java    | 190 +++++--
 .../parquet/proto/ProtoInputOutputFormatTest.java  | 120 +++++
 .../parquet/proto/ProtoSchemaConverterTest.java    | 219 +++++++-
 .../parquet/proto/ProtoWriteSupportTest.java       | 565 ++++++++++++++++++++-
 .../apache/parquet/proto/utils/WriteUsingMR.java   |  10 +-
 .../src/test/resources/TestProto3.proto            |   8 +
 .../src/test/resources/TestProtobuf.proto          |   8 +
 9 files changed, 1331 insertions(+), 84 deletions(-)

diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java
index 890f16c..979d78e 100644
--- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java
+++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java
@@ -24,12 +24,14 @@ import com.google.protobuf.Message;
 import com.twitter.elephantbird.util.Protobufs;
 import org.apache.parquet.column.Dictionary;
 import org.apache.parquet.io.InvalidRecordException;
+import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.Converter;
 import org.apache.parquet.io.api.GroupConverter;
 import org.apache.parquet.io.api.PrimitiveConverter;
 import org.apache.parquet.schema.GroupType;
 import org.apache.parquet.schema.IncompatibleSchemaModificationException;
+import org.apache.parquet.schema.OriginalType;
 import org.apache.parquet.schema.Type;
 
 import java.util.HashMap;
@@ -126,10 +128,15 @@ class ProtoMessageConverter extends GroupConverter {
       };
     }
 
+    if (OriginalType.LIST == parquetType.getOriginalType()) {
+      return new ListConverter(parentBuilder, fieldDescriptor, parquetType);
+    }
+    if (OriginalType.MAP == parquetType.getOriginalType()) {
+      return new MapConverter(parentBuilder, fieldDescriptor, parquetType);
+    }
     return newScalarConverter(parent, parentBuilder, fieldDescriptor, parquetType);
   }
 
-
   private Converter newScalarConverter(ParentValueContainer pvc, Message.Builder parentBuilder, Descriptors.FieldDescriptor fieldDescriptor, Type parquetType) {
 
     JavaType javaType = fieldDescriptor.getJavaType();
@@ -342,4 +349,121 @@ class ProtoMessageConverter extends GroupConverter {
     }
 
   }
+
+  /**
+   * This class unwraps the additional LIST wrapper and makes it possible to read the underlying data and then convert
+   * it to protobuf.
+   * <p>
+   * Consider the following protobuf schema:
+   * message SimpleList {
+   *   repeated int64 first_array = 1;
+   * }
+   * <p>
+   * A LIST wrapper is created in parquet for the above mentioned protobuf schema:
+   * message SimpleList {
+   *   optional group first_array (LIST) = 1 {
+   *     repeated group list {
+   *         optional int32 element;
+   *     }
+   *   }
+   * }
+   * <p>
+   * The LIST wrappers are used by 3rd party tools, such as Hive, to read parquet arrays. The wrapper contains
+   * a repeated group named 'list', itself containing only one field called 'element' of the type of the repeated
+   * object (can be a primitive as in this example or a group in case of a repeated message in protobuf).
+   */
+  final class ListConverter extends GroupConverter {
+    private final Converter converter;
+
+    public ListConverter(Message.Builder parentBuilder, Descriptors.FieldDescriptor fieldDescriptor, Type parquetType) {
+      OriginalType originalType = parquetType.getOriginalType();
+      if (originalType != OriginalType.LIST || parquetType.isPrimitive()) {
+        throw new ParquetDecodingException("Expected LIST wrapper. Found: " + originalType + " instead.");
+      }
+
+      GroupType rootWrapperType = parquetType.asGroupType();
+      if (!rootWrapperType.containsField("list") || rootWrapperType.getType("list").isPrimitive()) {
+        throw new ParquetDecodingException("Expected repeated 'list' group inside LIST wrapperr but got: " + rootWrapperType);
+      }
+
+      GroupType listType = rootWrapperType.getType("list").asGroupType();
+      if (!listType.containsField("element")) {
+        throw new ParquetDecodingException("Expected 'element' inside repeated list group but got: " + listType);
+      }
+
+      Type elementType = listType.getType("element");
+      converter = newMessageConverter(parentBuilder, fieldDescriptor, elementType);
+    }
+
+    @Override
+    public Converter getConverter(int fieldIndex) {
+      if (fieldIndex > 0) {
+        throw new ParquetDecodingException("Unexpected multiple fields in the LIST wrapper");
+      }
+
+      return new GroupConverter() {
+        @Override
+        public Converter getConverter(int fieldIndex) {
+          return converter;
+        }
+
+        @Override
+        public void start() {
+
+        }
+
+        @Override
+        public void end() {
+
+        }
+      };
+    }
+
+    @Override
+    public void start() {
+
+    }
+
+    @Override
+    public void end() {
+
+    }
+  }
+
+
+  final class MapConverter extends GroupConverter {
+    private final Converter converter;
+
+    public MapConverter(Message.Builder parentBuilder, Descriptors.FieldDescriptor fieldDescriptor, Type parquetType) {
+      OriginalType originalType = parquetType.getOriginalType();
+      if (originalType != OriginalType.MAP) {
+        throw new ParquetDecodingException("Expected MAP wrapper. Found: " + originalType + " instead.");
+      }
+
+      Type parquetSchema;
+      if (parquetType.asGroupType().containsField("key_value")){
+        parquetSchema = parquetType.asGroupType().getType("key_value");
+      } else {
+        throw new ParquetDecodingException("Expected map but got: " + parquetType);
+      }
+
+      converter = newMessageConverter(parentBuilder, fieldDescriptor, parquetSchema);
+    }
+
+    @Override
+    public Converter getConverter(int fieldIndex) {
+      if (fieldIndex > 0) {
+        throw new ParquetDecodingException("Unexpected multiple fields in the MAP wrapper");
+      }
+      return converter;
+    }
+
+    @Override
+    public void start() {
+    }
+
+    @Override
+    public void end() {
+    }
+  }
 }
diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoSchemaConverter.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoSchemaConverter.java
index 64668c0..0e1aa20 100644
--- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoSchemaConverter.java
+++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoSchemaConverter.java
@@ -1,4 +1,4 @@
-/* 
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -18,36 +18,49 @@
  */
 package org.apache.parquet.proto;
 
-import static org.apache.parquet.schema.OriginalType.ENUM;
-import static org.apache.parquet.schema.OriginalType.UTF8;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BOOLEAN;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
-import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
-
-import java.util.List;
-
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.Descriptors.FieldDescriptor;
+import com.google.protobuf.Descriptors.FieldDescriptor.JavaType;
+import com.google.protobuf.Message;
+import com.twitter.elephantbird.util.Protobufs;
 import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import org.apache.parquet.schema.Type;
 import org.apache.parquet.schema.Types;
 import org.apache.parquet.schema.Types.Builder;
 import org.apache.parquet.schema.Types.GroupBuilder;
-
-import com.google.protobuf.Descriptors;
-import com.google.protobuf.Descriptors.FieldDescriptor.JavaType;
-import com.google.protobuf.Message;
-import com.twitter.elephantbird.util.Protobufs;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+
+import static org.apache.parquet.schema.OriginalType.ENUM;
+import static org.apache.parquet.schema.OriginalType.UTF8;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.*;
+
 /**
  * Converts a Protocol Buffer Descriptor into a Parquet schema.
  */
 public class ProtoSchemaConverter {
 
   private static final Logger LOG = LoggerFactory.getLogger(ProtoSchemaConverter.class);
+  private final boolean parquetSpecsCompliant;
+
+  public ProtoSchemaConverter() {
+    this(false);
+  }
+
+  /**
+   * Instanciate a schema converter to get the parquet schema corresponding to protobuf classes.
+   * @param parquetSpecsCompliant   If set to false, the parquet schema generated will be using the old
+   *                                schema style (prior to PARQUET-968) to provide backward-compatibility
+   *                                but which does not use LIST and MAP wrappers around collections as required
+   *                                by the parquet specifications. If set to true, specs compliant schemas are used.
+   */
+  public ProtoSchemaConverter(boolean parquetSpecsCompliant) {
+    this.parquetSpecsCompliant = parquetSpecsCompliant;
+  }
 
   public MessageType convert(Class<? extends Message> protobufClass) {
     LOG.debug("Converting protocol buffer class \"" + protobufClass + "\" to parquet schema.");
@@ -60,8 +73,8 @@ public class ProtoSchemaConverter {
   }
 
   /* Iterates over list of fields. **/
-  private <T> GroupBuilder<T> convertFields(GroupBuilder<T> groupBuilder, List<Descriptors.FieldDescriptor> fieldDescriptors) {
-    for (Descriptors.FieldDescriptor fieldDescriptor : fieldDescriptors) {
+  private <T> GroupBuilder<T> convertFields(GroupBuilder<T> groupBuilder, List<FieldDescriptor> fieldDescriptors) {
+    for (FieldDescriptor fieldDescriptor : fieldDescriptors) {
       groupBuilder =
           addField(fieldDescriptor, groupBuilder)
           .id(fieldDescriptor.getNumber())
@@ -70,7 +83,7 @@ public class ProtoSchemaConverter {
     return groupBuilder;
   }
 
-  private Type.Repetition getRepetition(Descriptors.FieldDescriptor descriptor) {
+  private Type.Repetition getRepetition(FieldDescriptor descriptor) {
     if (descriptor.isRequired()) {
       return Type.Repetition.REQUIRED;
     } else if (descriptor.isRepeated()) {
@@ -80,26 +93,110 @@ public class ProtoSchemaConverter {
     }
   }
 
-  private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addField(Descriptors.FieldDescriptor descriptor, GroupBuilder<T> builder) {
-    Type.Repetition repetition = getRepetition(descriptor);
-    JavaType javaType = descriptor.getJavaType();
+  private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addField(FieldDescriptor descriptor, final GroupBuilder<T> builder) {
+    if (descriptor.getJavaType() == JavaType.MESSAGE) {
+      return addMessageField(descriptor, builder);
+    }
+
+    ParquetType parquetType = getParquetType(descriptor);
+    if (descriptor.isRepeated() && parquetSpecsCompliant) {
+      // the old schema style did not include the LIST wrapper around repeated fields
+      return addRepeatedPrimitive(descriptor, parquetType.primitiveType, parquetType.originalType, builder);
+    }
+
+    return builder.primitive(parquetType.primitiveType, getRepetition(descriptor)).as(parquetType.originalType);
+  }
+
+  private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addRepeatedPrimitive(FieldDescriptor descriptor,
+                                                                                                   PrimitiveTypeName primitiveType,
+                                                                                                   OriginalType originalType,
+                                                                                                   final GroupBuilder<T> builder) {
+    return builder
+        .group(Type.Repetition.OPTIONAL).as(OriginalType.LIST)
+          .group(Type.Repetition.REPEATED)
+            .primitive(primitiveType, Type.Repetition.REQUIRED).as(originalType)
+          .named("element")
+        .named("list");
+  }
+
+  private <T> GroupBuilder<GroupBuilder<T>> addRepeatedMessage(FieldDescriptor descriptor, GroupBuilder<T> builder) {
+    GroupBuilder<GroupBuilder<GroupBuilder<GroupBuilder<T>>>> result =
+      builder
+        .group(Type.Repetition.OPTIONAL).as(OriginalType.LIST)
+        .group(Type.Repetition.REPEATED)
+        .group(Type.Repetition.OPTIONAL);
+
+    convertFields(result, descriptor.getMessageType().getFields());
+
+    return result.named("element").named("list");
+  }
+
+  private <T> GroupBuilder<GroupBuilder<T>> addMessageField(FieldDescriptor descriptor, final GroupBuilder<T> builder) {
+    if (descriptor.isMapField() && parquetSpecsCompliant) {
+      // the old schema style did not include the MAP wrapper around map groups
+      return addMapField(descriptor, builder);
+    }
+    if (descriptor.isRepeated() && parquetSpecsCompliant) {
+      // the old schema style did not include the LIST wrapper around repeated messages
+      return addRepeatedMessage(descriptor, builder);
+    }
+
+    // Plain message
+    GroupBuilder<GroupBuilder<T>> group = builder.group(getRepetition(descriptor));
+    convertFields(group, descriptor.getMessageType().getFields());
+    return group;
+  }
+
+  private <T> GroupBuilder<GroupBuilder<T>> addMapField(FieldDescriptor descriptor, final GroupBuilder<T> builder) {
+    List<FieldDescriptor> fields = descriptor.getMessageType().getFields();
+    if (fields.size() != 2) {
+      throw new UnsupportedOperationException("Expected two fields for the map (key/value), but got: " + fields);
+    }
+
+    ParquetType mapKeyParquetType = getParquetType(fields.get(0));
+
+    GroupBuilder<GroupBuilder<GroupBuilder<T>>> group = builder
+      .group(Type.Repetition.OPTIONAL).as(OriginalType.MAP) // only optional maps are allowed in Proto3
+      .group(Type.Repetition.REPEATED) // key_value wrapper
+      .primitive(mapKeyParquetType.primitiveType, Type.Repetition.REQUIRED).as(mapKeyParquetType.originalType).named("key");
+
+    return addField(fields.get(1), group).named("value")
+      .named("key_value");
+  }
+
+  private ParquetType getParquetType(FieldDescriptor fieldDescriptor) {
+
+    JavaType javaType = fieldDescriptor.getJavaType();
     switch (javaType) {
-      case BOOLEAN: return builder.primitive(BOOLEAN, repetition);
-      case INT: return builder.primitive(INT32, repetition);
-      case LONG: return builder.primitive(INT64, repetition);
-      case FLOAT: return builder.primitive(FLOAT, repetition);
-      case DOUBLE: return builder.primitive(DOUBLE, repetition);
-      case BYTE_STRING: return builder.primitive(BINARY, repetition);
-      case STRING: return builder.primitive(BINARY, repetition).as(UTF8);
-      case MESSAGE: {
-        GroupBuilder<GroupBuilder<T>> group = builder.group(repetition);
-        convertFields(group, descriptor.getMessageType().getFields());
-        return group;
-      }
-      case ENUM: return builder.primitive(BINARY, repetition).as(ENUM);
+      case INT: return ParquetType.of(INT32);
+      case LONG: return ParquetType.of(INT64);
+      case DOUBLE: return ParquetType.of(DOUBLE);
+      case BOOLEAN: return ParquetType.of(BOOLEAN);
+      case FLOAT: return ParquetType.of(FLOAT);
+      case STRING: return ParquetType.of(BINARY, UTF8);
+      case ENUM: return ParquetType.of(BINARY, ENUM);
+      case BYTE_STRING: return ParquetType.of(BINARY);
       default:
         throw new UnsupportedOperationException("Cannot convert Protocol Buffer: unknown type " + javaType);
     }
   }
 
+  private static class ParquetType {
+    PrimitiveTypeName primitiveType;
+    OriginalType originalType;
+
+    private ParquetType(PrimitiveTypeName primitiveType, OriginalType originalType) {
+      this.primitiveType = primitiveType;
+      this.originalType = originalType;
+    }
+
+    public static ParquetType of(PrimitiveTypeName primitiveType, OriginalType originalType) {
+      return new ParquetType(primitiveType, originalType);
+    }
+
+    public static ParquetType of(PrimitiveTypeName primitiveType) {
+      return of(primitiveType, null);
+    }
+  }
+
 }
diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoWriteSupport.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoWriteSupport.java
index 19b0706..59c236f 100644
--- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoWriteSupport.java
+++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoWriteSupport.java
@@ -18,12 +18,9 @@
  */
 package org.apache.parquet.proto;
 
-import com.google.protobuf.ByteString;
-import com.google.protobuf.DescriptorProtos;
-import com.google.protobuf.Descriptors;
-import com.google.protobuf.Message;
-import com.google.protobuf.MessageOrBuilder;
-import com.google.protobuf.TextFormat;
+import com.google.protobuf.*;
+import com.google.protobuf.Descriptors.Descriptor;
+import com.google.protobuf.Descriptors.FieldDescriptor;
 import com.twitter.elephantbird.util.Protobufs;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.parquet.hadoop.BadConfigurationException;
@@ -31,14 +28,13 @@ import org.apache.parquet.hadoop.api.WriteSupport;
 import org.apache.parquet.io.InvalidRecordException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.RecordConsumer;
-import org.apache.parquet.schema.GroupType;
-import org.apache.parquet.schema.IncompatibleSchemaModificationException;
-import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.*;
 import org.apache.parquet.schema.Type;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.Array;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -50,7 +46,13 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
 
   private static final Logger LOG = LoggerFactory.getLogger(ProtoWriteSupport.class);
   public static final String PB_CLASS_WRITE = "parquet.proto.writeClass";
+  // PARQUET-968 introduces changes to allow writing specs compliant schemas with parquet-protobuf.
+  // In the past, collection were not written using the LIST and MAP wrappers and thus were not compliant
+  // with the parquet specs. This flag, is set to true, allows to write using spec compliant schemas
+  // but is set to false by default to keep backward compatibility
+  public static final String PB_SPECS_COMPLIANT_WRITE = "parquet.proto.writeSpecsCompliant";
 
+  private boolean writeSpecsCompliant = false;
   private RecordConsumer recordConsumer;
   private Class<? extends Message> protoMessage;
   private MessageWriter messageWriter;
@@ -72,6 +74,16 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
   }
 
   /**
+   * Make parquet-protobuf use the LIST and MAP wrappers for collections. Set to false if you need backward
+   * compatibility with parquet before PARQUET-968 (1.9.0 and older).
+   * @param configuration           The hadoop configuration
+   * @param writeSpecsCompliant     If set to true, the old schema style will be used (without wrappers).
+   */
+  public static void setWriteSpecsCompliant(Configuration configuration, boolean writeSpecsCompliant) {
+    configuration.setBoolean(PB_SPECS_COMPLIANT_WRITE, writeSpecsCompliant);
+  }
+
+  /**
    * Writes Protocol buffer to parquet file.
    * @param record instance of Message.Builder or Message.
    * */
@@ -108,8 +120,9 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
       }
     }
 
-    MessageType rootSchema = new ProtoSchemaConverter().convert(protoMessage);
-    Descriptors.Descriptor messageDescriptor = Protobufs.getMessageDescriptor(protoMessage);
+    writeSpecsCompliant = configuration.getBoolean(PB_SPECS_COMPLIANT_WRITE, writeSpecsCompliant);
+    MessageType rootSchema = new ProtoSchemaConverter(writeSpecsCompliant).convert(protoMessage);
+    Descriptor messageDescriptor = Protobufs.getMessageDescriptor(protoMessage);
     validatedMapping(messageDescriptor, rootSchema);
 
     this.messageWriter = new MessageWriter(messageDescriptor, rootSchema);
@@ -117,6 +130,7 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     Map<String, String> extraMetaData = new HashMap<String, String>();
     extraMetaData.put(ProtoReadSupport.PB_CLASS, protoMessage.getName());
     extraMetaData.put(ProtoReadSupport.PB_DESCRIPTOR, serializeDescriptor(protoMessage));
+    extraMetaData.put(PB_SPECS_COMPLIANT_WRITE, String.valueOf(writeSpecsCompliant));
     return new WriteContext(rootSchema, extraMetaData);
   }
 
@@ -152,17 +166,21 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     final FieldWriter[] fieldWriters;
 
     @SuppressWarnings("unchecked")
-    MessageWriter(Descriptors.Descriptor descriptor, GroupType schema) {
-      List<Descriptors.FieldDescriptor> fields = descriptor.getFields();
+    MessageWriter(Descriptor descriptor, GroupType schema) {
+      List<FieldDescriptor> fields = descriptor.getFields();
       fieldWriters = (FieldWriter[]) Array.newInstance(FieldWriter.class, fields.size());
 
-      for (Descriptors.FieldDescriptor fieldDescriptor: fields) {
+      for (FieldDescriptor fieldDescriptor: fields) {
         String name = fieldDescriptor.getName();
         Type type = schema.getType(name);
         FieldWriter writer = createWriter(fieldDescriptor, type);
 
-        if(fieldDescriptor.isRepeated()) {
-         writer = new ArrayWriter(writer);
+        if(writeSpecsCompliant && fieldDescriptor.isRepeated() && !fieldDescriptor.isMapField()) {
+          writer = new ArrayWriter(writer);
+        }
+        else if (!writeSpecsCompliant && fieldDescriptor.isRepeated()) {
+          // the old schemas style used to write maps as repeated fields instead of wrapping them in a LIST
+          writer = new RepeatedWriter(writer);
         }
 
         writer.setFieldName(name);
@@ -172,11 +190,11 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
       }
     }
 
-    private FieldWriter createWriter(Descriptors.FieldDescriptor fieldDescriptor, Type type) {
+    private FieldWriter createWriter(FieldDescriptor fieldDescriptor, Type type) {
 
       switch (fieldDescriptor.getJavaType()) {
         case STRING: return new StringWriter() ;
-        case MESSAGE: return new MessageWriter(fieldDescriptor.getMessageType(), type.asGroupType());
+        case MESSAGE: return createMessageWriter(fieldDescriptor, type);
         case INT: return new IntWriter();
         case LONG: return new LongWriter();
         case FLOAT: return new FloatWriter();
@@ -189,6 +207,47 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
       return unknownType(fieldDescriptor);//should not be executed, always throws exception.
     }
 
+    private FieldWriter createMessageWriter(FieldDescriptor fieldDescriptor, Type type) {
+      if (fieldDescriptor.isMapField() && writeSpecsCompliant) {
+        return createMapWriter(fieldDescriptor, type);
+      }
+
+      return new MessageWriter(fieldDescriptor.getMessageType(), getGroupType(type));
+    }
+
+    private GroupType getGroupType(Type type) {
+      if (type.getOriginalType() == OriginalType.LIST) {
+        return type.asGroupType().getType("list").asGroupType().getType("element").asGroupType();
+      }
+
+      if (type.getOriginalType() == OriginalType.MAP) {
+        return type.asGroupType().getType("key_value").asGroupType().getType("value").asGroupType();
+      }
+
+      return type.asGroupType();
+    }
+
+    private MapWriter createMapWriter(FieldDescriptor fieldDescriptor, Type type) {
+      List<FieldDescriptor> fields = fieldDescriptor.getMessageType().getFields();
+      if (fields.size() != 2) {
+        throw new UnsupportedOperationException("Expected two fields for the map (key/value), but got: " + fields);
+      }
+
+      // KeyFieldWriter
+      FieldDescriptor keyProtoField = fields.get(0);
+      FieldWriter keyWriter = createWriter(keyProtoField, type);
+      keyWriter.setFieldName(keyProtoField.getName());
+      keyWriter.setIndex(0);
+
+      // ValueFieldWriter
+      FieldDescriptor valueProtoField = fields.get(1);
+      FieldWriter valueWriter = createWriter(valueProtoField, type);
+      valueWriter.setFieldName(valueProtoField.getName());
+      valueWriter.setIndex(1);
+
+      return new MapWriter(keyWriter, valueWriter);
+    }
+
     /** Writes top level message. It cannot call startGroup() */
     void writeTopLevelMessage(Object value) {
       writeAllFields((MessageOrBuilder) value);
@@ -206,18 +265,16 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     @Override
     final void writeField(Object value) {
       recordConsumer.startField(fieldName, index);
-      recordConsumer.startGroup();
-      writeAllFields((MessageOrBuilder) value);
-      recordConsumer.endGroup();
+      writeRawValue(value);
       recordConsumer.endField(fieldName, index);
     }
 
     private void writeAllFields(MessageOrBuilder pb) {
       //returns changed fields with values. Map is ordered by id.
-      Map<Descriptors.FieldDescriptor, Object> changedPbFields = pb.getAllFields();
+      Map<FieldDescriptor, Object> changedPbFields = pb.getAllFields();
 
-      for (Map.Entry<Descriptors.FieldDescriptor, Object> entry : changedPbFields.entrySet()) {
-        Descriptors.FieldDescriptor fieldDescriptor = entry.getKey();
+      for (Map.Entry<FieldDescriptor, Object> entry : changedPbFields.entrySet()) {
+        FieldDescriptor fieldDescriptor = entry.getKey();
 
         if(fieldDescriptor.isExtension()) {
           // Field index of an extension field might overlap with a base field.
@@ -246,6 +303,45 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     @Override
     final void writeField(Object value) {
       recordConsumer.startField(fieldName, index);
+      recordConsumer.startGroup();
+      List<?> list = (List<?>) value;
+
+      recordConsumer.startField("list", 0); // This is the wrapper group for the array field
+      for (Object listEntry: list) {
+        recordConsumer.startGroup();
+        recordConsumer.startField("element", 0); // This is the mandatory inner field
+
+        fieldWriter.writeRawValue(listEntry);
+
+        recordConsumer.endField("element", 0);
+        recordConsumer.endGroup();
+      }
+      recordConsumer.endField("list", 0);
+
+      recordConsumer.endGroup();
+      recordConsumer.endField(fieldName, index);
+    }
+  }
+
+  /**
+   * The RepeatedWriter is used to write collections (lists and maps) using the old style (without LIST and MAP
+   * wrappers).
+   */
+  class RepeatedWriter extends FieldWriter {
+    final FieldWriter fieldWriter;
+
+    RepeatedWriter(FieldWriter fieldWriter) {
+      this.fieldWriter = fieldWriter;
+    }
+
+    @Override
+    final void writeRawValue(Object value) {
+      throw new UnsupportedOperationException("Array has no raw value");
+    }
+
+    @Override
+    final void writeField(Object value) {
+      recordConsumer.startField(fieldName, index);
       List<?> list = (List<?>) value;
 
       for (Object listEntry: list) {
@@ -257,10 +353,10 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
   }
 
   /** validates mapping between protobuffer fields and parquet fields.*/
-  private void validatedMapping(Descriptors.Descriptor descriptor, GroupType parquetSchema) {
-    List<Descriptors.FieldDescriptor> allFields = descriptor.getFields();
+  private void validatedMapping(Descriptor descriptor, GroupType parquetSchema) {
+    List<FieldDescriptor> allFields = descriptor.getFields();
 
-    for (Descriptors.FieldDescriptor fieldDescriptor: allFields) {
+    for (FieldDescriptor fieldDescriptor: allFields) {
       String fieldName = fieldDescriptor.getName();
       int fieldIndex = fieldDescriptor.getIndex();
       int parquetIndex = parquetSchema.getFieldIndex(fieldName);
@@ -295,6 +391,41 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     }
   }
 
+  class MapWriter extends FieldWriter {
+
+    private final FieldWriter keyWriter;
+    private final FieldWriter valueWriter;
+
+    public MapWriter(FieldWriter keyWriter, FieldWriter valueWriter) {
+      super();
+      this.keyWriter = keyWriter;
+      this.valueWriter = valueWriter;
+    }
+
+    @Override
+    final void writeRawValue(Object value) {
+      recordConsumer.startGroup();
+
+      recordConsumer.startField("key_value", 0); // This is the wrapper group for the map field
+      for (Message msg : (Collection<Message>) value) {
+        recordConsumer.startGroup();
+
+        final Descriptor descriptorForType = msg.getDescriptorForType();
+        final FieldDescriptor keyDesc = descriptorForType.findFieldByName("key");
+        final FieldDescriptor valueDesc = descriptorForType.findFieldByName("value");
+
+        keyWriter.writeField(msg.getField(keyDesc));
+        valueWriter.writeField(msg.getField(valueDesc));
+
+        recordConsumer.endGroup();
+      }
+
+      recordConsumer.endField("key_value", 0);
+
+      recordConsumer.endGroup();
+    }
+  }
+
   class FloatWriter extends FieldWriter {
     @Override
     final void writeRawValue(Object value) {
@@ -333,7 +464,7 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
     }
   }
 
-  private FieldWriter unknownType(Descriptors.FieldDescriptor fieldDescriptor) {
+  private FieldWriter unknownType(FieldDescriptor fieldDescriptor) {
     String exceptionMsg = "Unknown type with descriptor \"" + fieldDescriptor
             + "\" and type \"" + fieldDescriptor.getJavaType() + "\".";
     throw new InvalidRecordException(exceptionMsg);
@@ -341,9 +472,8 @@ public class ProtoWriteSupport<T extends MessageOrBuilder> extends WriteSupport<
 
   /** Returns message descriptor as JSON String*/
   private String serializeDescriptor(Class<? extends Message> protoClass) {
-    Descriptors.Descriptor descriptor = Protobufs.getMessageDescriptor(protoClass);
+    Descriptor descriptor = Protobufs.getMessageDescriptor(protoClass);
     DescriptorProtos.DescriptorProto asProto = descriptor.toProto();
     return TextFormat.printToString(asProto);
   }
-
 }
diff --git a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoInputOutputFormatTest.java b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoInputOutputFormatTest.java
index 6c01d7b..5544dc6 100644
--- a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoInputOutputFormatTest.java
+++ b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoInputOutputFormatTest.java
@@ -19,6 +19,7 @@
 package org.apache.parquet.proto;
 
 import com.google.protobuf.Message;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.parquet.proto.test.TestProto3;
 import org.apache.parquet.proto.test.TestProtobuf;
@@ -193,6 +194,125 @@ public class ProtoInputOutputFormatTest {
     assertEquals("writtenString", stringValue);
   }
 
+  @Test
+  public void testRepeatedIntMessageClass() throws Exception {
+    TestProtobuf.RepeatedIntMessage msgEmpty = TestProtobuf.RepeatedIntMessage.newBuilder().build();
+    TestProtobuf.RepeatedIntMessage msgNonEmpty = TestProtobuf.RepeatedIntMessage.newBuilder()
+      .addRepeatedInt(1).addRepeatedInt(2)
+      .build();
+
+    Path outputPath = new WriteUsingMR().write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.RepeatedIntMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
+  @Test
+  public void testRepeatedIntMessageClassSchemaCompliant() throws Exception {
+    TestProtobuf.RepeatedIntMessage msgEmpty = TestProtobuf.RepeatedIntMessage.newBuilder().build();
+    TestProtobuf.RepeatedIntMessage msgNonEmpty = TestProtobuf.RepeatedIntMessage.newBuilder()
+      .addRepeatedInt(1).addRepeatedInt(2)
+      .build();
+
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+
+    Path outputPath = new WriteUsingMR(conf).write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.RepeatedIntMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
+  @Test
+  public void testMapIntMessageClass() throws Exception {
+    TestProtobuf.MapIntMessage msgEmpty = TestProtobuf.MapIntMessage.newBuilder().build();
+    TestProtobuf.MapIntMessage msgNonEmpty = TestProtobuf.MapIntMessage.newBuilder()
+      .putMapInt(1, 123).putMapInt(2, 234)
+      .build();
+
+    Path outputPath = new WriteUsingMR().write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.MapIntMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
+  @Test
+  public void testMapIntMessageClassSchemaCompliant() throws Exception {
+    TestProtobuf.MapIntMessage msgEmpty = TestProtobuf.MapIntMessage.newBuilder().build();
+    TestProtobuf.MapIntMessage msgNonEmpty = TestProtobuf.MapIntMessage.newBuilder()
+      .putMapInt(1, 123).putMapInt(2, 234)
+      .build();
+
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+
+    Path outputPath = new WriteUsingMR(conf).write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.MapIntMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
+  @Test
+  public void testRepeatedInnerMessageClass() throws Exception {
+    TestProtobuf.RepeatedInnerMessage msgEmpty = TestProtobuf.RepeatedInnerMessage.newBuilder().build();
+    TestProtobuf.RepeatedInnerMessage msgNonEmpty = TestProtobuf.RepeatedInnerMessage.newBuilder()
+      .addRepeatedInnerMessage(TestProtobuf.InnerMessage.newBuilder().setOne("one").build())
+      .addRepeatedInnerMessage(TestProtobuf.InnerMessage.newBuilder().setTwo("two").build())
+      .build();
+
+    Path outputPath = new WriteUsingMR().write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.RepeatedInnerMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
+  @Test
+  public void testRepeatedInnerMessageClassSchemaCompliant() throws Exception {
+    TestProtobuf.RepeatedInnerMessage msgEmpty = TestProtobuf.RepeatedInnerMessage.newBuilder().build();
+    TestProtobuf.RepeatedInnerMessage msgNonEmpty = TestProtobuf.RepeatedInnerMessage.newBuilder()
+      .addRepeatedInnerMessage(TestProtobuf.InnerMessage.newBuilder().setOne("one").build())
+      .addRepeatedInnerMessage(TestProtobuf.InnerMessage.newBuilder().setTwo("two").build())
+      .build();
+
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+
+    Path outputPath = new WriteUsingMR(conf).write(msgEmpty, msgNonEmpty);
+    ReadUsingMR readUsingMR = new ReadUsingMR();
+    String customClass = TestProtobuf.RepeatedInnerMessage.class.getName();
+    ProtoReadSupport.setProtobufClass(readUsingMR.getConfiguration(), customClass);
+    List<Message> result = readUsingMR.read(outputPath);
+
+    assertEquals(2, result.size());
+    assertEquals(msgEmpty, result.get(0));
+    assertEquals(msgNonEmpty, result.get(1));
+  }
+
   /**
    * Runs job that writes input to file and then job reading data back.
    */
diff --git a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoSchemaConverterTest.java b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoSchemaConverterTest.java
index 6f5ff53..4ca82ac 100644
--- a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoSchemaConverterTest.java
+++ b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoSchemaConverterTest.java
@@ -32,14 +32,17 @@ public class ProtoSchemaConverterTest {
   /**
    * Converts given pbClass to parquet schema and compares it with expected parquet schema.
    */
-  private void testConversion(Class<? extends Message> pbClass, String parquetSchemaString) throws
+  private void testConversion(Class<? extends Message> pbClass, String parquetSchemaString, boolean parquetSpecsCompliant) throws
           Exception {
-    ProtoSchemaConverter protoSchemaConverter = new ProtoSchemaConverter();
+    ProtoSchemaConverter protoSchemaConverter = new ProtoSchemaConverter(parquetSpecsCompliant);
     MessageType schema = protoSchemaConverter.convert(pbClass);
     MessageType expectedMT = MessageTypeParser.parseMessageType(parquetSchemaString);
     assertEquals(expectedMT.toString(), schema.toString());
   }
 
+  private void testConversion(Class<? extends Message> pbClass, String parquetSchemaString) throws Exception {
+    testConversion(pbClass, parquetSchemaString, true);
+  }
 
   /**
    * Tests that all protocol buffer datatypes are converted to correct parquet datatypes.
@@ -103,10 +106,12 @@ public class ProtoSchemaConverterTest {
         "  optional binary optionalEnum (ENUM) = 18;" +
         "  optional int32 someInt32 = 19;" +
         "  optional binary someString (UTF8) = 20;" +
-        "  repeated group optionalMap = 21 {\n" +
-        "    optional int64 key = 1;\n" +
-        "    optional group value = 2 {\n" +
-        "      optional int32 someId = 3;\n" +
+        "  optional group optionalMap (MAP) = 21 {\n" +
+        "    repeated group key_value {\n" +
+        "      required int64 key;\n" +
+        "      optional group value {\n" +
+        "        optional int32 someId = 3;\n" +
+        "      }\n" +
         "    }\n" +
         "  }\n" +
         "}";
@@ -120,16 +125,24 @@ public class ProtoSchemaConverterTest {
       "message TestProtobuf.SchemaConverterRepetition {\n" +
         "  optional int32 optionalPrimitive = 1;\n" +
         "  required int32 requiredPrimitive = 2;\n" +
-        "  repeated int32 repeatedPrimitive = 3;\n" +
+        "  optional group repeatedPrimitive (LIST) = 3 {\n" +
+        "    repeated group list {\n" +
+        "      required int32 element;\n" +
+        "    }\n" +
+        "  }\n" +
         "  optional group optionalMessage = 7 {\n" +
         "    optional int32 someId = 3;\n" +
         "  }\n" +
-        "  required group requiredMessage = 8 {" +
+        "  required group requiredMessage = 8 {\n" +
         "    optional int32 someId= 3;\n" +
         "  }\n" +
-        "  repeated group repeatedMessage = 9 {" +
-        "    optional int32 someId = 3;\n" +
-        "  }\n" +
+        "  optional group repeatedMessage (LIST) = 9 {\n" +
+        "    repeated group list {\n" +
+        "      optional group element {\n" +
+        "        optional int32 someId = 3;\n" +
+        "      }\n" +
+        "    }\n" +
+        "  }" +
         "}";
 
     testConversion(TestProtobuf.SchemaConverterRepetition.class, expectedSchema);
@@ -140,15 +153,193 @@ public class ProtoSchemaConverterTest {
     String expectedSchema =
       "message TestProto3.SchemaConverterRepetition {\n" +
         "  optional int32 optionalPrimitive = 1;\n" +
-        "  repeated int32 repeatedPrimitive = 3;\n" +
+        "  optional group repeatedPrimitive (LIST) = 3 {\n" +
+        "    repeated group list {\n" +
+        "      required int32 element;\n" +
+        "    }\n" +
+        "  }\n" +
         "  optional group optionalMessage = 7 {\n" +
         "    optional int32 someId = 3;\n" +
         "  }\n" +
-        "  repeated group repeatedMessage = 9 {" +
-        "    optional int32 someId = 3;\n" +
+        "  optional group repeatedMessage (LIST) = 9 {\n" +
+        "    repeated group list {\n" +
+        "      optional group element {\n" +
+        "        optional int32 someId = 3;\n" +
+        "      }\n" +
+        "    }\n" +
         "  }\n" +
         "}";
 
     testConversion(TestProto3.SchemaConverterRepetition.class, expectedSchema);
   }
+
+  @Test
+  public void testConvertRepeatedIntMessage() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.RepeatedIntMessage {\n" +
+        "  optional group repeatedInt (LIST) = 1 {\n" +
+        "    repeated group list {\n" +
+        "      required int32 element;\n" +
+        "      }\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProtobuf.RepeatedIntMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testConvertRepeatedIntMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.RepeatedIntMessage {\n" +
+        "  repeated int32 repeatedInt = 1;\n" +
+        "}";
+
+    testConversion(TestProtobuf.RepeatedIntMessage.class, expectedSchema, false);
+  }
+
+  @Test
+  public void testProto3ConvertRepeatedIntMessage() throws Exception {
+    String expectedSchema =
+      "message TestProto3.RepeatedIntMessage {\n" +
+        "  optional group repeatedInt (LIST) = 1 {\n" +
+        "    repeated group list {\n" +
+        "      required int32 element;\n" +
+        "      }\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProto3.RepeatedIntMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testProto3ConvertRepeatedIntMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProto3.RepeatedIntMessage {\n" +
+        "  repeated int32 repeatedInt = 1;\n" +
+        "}";
+
+    testConversion(TestProto3.RepeatedIntMessage.class, expectedSchema, false);
+  }
+
+  @Test
+  public void testConvertRepeatedInnerMessage() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.RepeatedInnerMessage {\n" +
+        "  optional group repeatedInnerMessage (LIST) = 1 {\n" +
+        "    repeated group list {\n" +
+        "      optional group element {\n" +
+        "        optional binary one (UTF8) = 1;\n" +
+        "        optional binary two (UTF8) = 2;\n" +
+        "        optional binary three (UTF8) = 3;\n" +
+        "      }\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProtobuf.RepeatedInnerMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testConvertRepeatedInnerMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.RepeatedInnerMessage {\n" +
+        "  repeated group repeatedInnerMessage = 1 {\n" +
+        "    optional binary one (UTF8) = 1;\n" +
+        "    optional binary two (UTF8) = 2;\n" +
+        "    optional binary three (UTF8) = 3;\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProtobuf.RepeatedInnerMessage.class, expectedSchema, false);
+  }
+
+  @Test
+  public void testProto3ConvertRepeatedInnerMessage() throws Exception {
+    String expectedSchema =
+      "message TestProto3.RepeatedInnerMessage {\n" +
+        "  optional group repeatedInnerMessage (LIST) = 1 {\n" +
+        "    repeated group list {\n" +
+        "      optional group element {\n" +
+        "        optional binary one (UTF8) = 1;\n" +
+        "        optional binary two (UTF8) = 2;\n" +
+        "        optional binary three (UTF8) = 3;\n" +
+        "      }\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProto3.RepeatedInnerMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testProto3ConvertRepeatedInnerMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProto3.RepeatedInnerMessage {\n" +
+        "  repeated group repeatedInnerMessage = 1 {\n" +
+        "    optional binary one (UTF8) = 1;\n" +
+        "    optional binary two (UTF8) = 2;\n" +
+        "    optional binary three (UTF8) = 3;\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProto3.RepeatedInnerMessage.class, expectedSchema, false);
+  }
+
+  @Test
+  public void testConvertMapIntMessage() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.MapIntMessage {\n" +
+        "  optional group mapInt (MAP) = 1 {\n" +
+        "    repeated group key_value {\n" +
+        "      required int32 key;\n" +
+        "      optional int32 value;\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProtobuf.MapIntMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testConvertMapIntMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProtobuf.MapIntMessage {\n" +
+        "  repeated group mapInt = 1 {\n" +
+        "    optional int32 key = 1;\n" +
+        "    optional int32 value = 2;\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProtobuf.MapIntMessage.class, expectedSchema, false);
+  }
+
+  @Test
+  public void testProto3ConvertMapIntMessage() throws Exception {
+    String expectedSchema =
+      "message TestProto3.MapIntMessage {\n" +
+        "  optional group mapInt (MAP) = 1 {\n" +
+        "    repeated group key_value {\n" +
+        "      required int32 key;\n" +
+        "      optional int32 value;\n" +
+        "    }\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProto3.MapIntMessage.class, expectedSchema);
+  }
+
+  @Test
+  public void testProto3ConvertMapIntMessageNonSpecsCompliant() throws Exception {
+    String expectedSchema =
+      "message TestProto3.MapIntMessage {\n" +
+        "  repeated group mapInt = 1 {\n" +
+        "    optional int32 key = 1;\n" +
+        "    optional int32 value = 2;\n" +
+        "  }\n" +
+        "}";
+
+    testConversion(TestProto3.MapIntMessage.class, expectedSchema, false);
+  }
 }
diff --git a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoWriteSupportTest.java b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoWriteSupportTest.java
index b937618..f71229c 100644
--- a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoWriteSupportTest.java
+++ b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoWriteSupportTest.java
@@ -31,8 +31,12 @@ import org.apache.parquet.proto.test.TestProtobuf;
 public class ProtoWriteSupportTest {
 
   private <T extends Message> ProtoWriteSupport<T> createReadConsumerInstance(Class<T> cls, RecordConsumer readConsumerMock) {
+    return createReadConsumerInstance(cls, readConsumerMock, new Configuration());
+  }
+
+  private <T extends Message> ProtoWriteSupport<T> createReadConsumerInstance(Class<T> cls, RecordConsumer readConsumerMock, Configuration conf) {
     ProtoWriteSupport support = new ProtoWriteSupport(cls);
-    support.init(new Configuration());
+    support.init(conf);
     support.prepareForWrite(readConsumerMock);
     return support;
   }
@@ -80,6 +84,45 @@ public class ProtoWriteSupportTest {
   }
 
   @Test
+  public void testRepeatedIntMessageSpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.RepeatedIntMessage.Builder msg = TestProtobuf.RepeatedIntMessage.newBuilder();
+    msg.addRepeatedInt(1323);
+    msg.addRepeatedInt(54469);
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("repeatedInt", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).addInteger(1323);
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).addInteger(54469);
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("repeatedInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
   public void testRepeatedIntMessage() throws Exception {
     RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
     ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock);
@@ -102,6 +145,79 @@ public class ProtoWriteSupportTest {
   }
 
   @Test
+  public void testRepeatedIntMessageEmptySpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.RepeatedIntMessage.Builder msg = TestProtobuf.RepeatedIntMessage.newBuilder();
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testRepeatedIntMessageEmpty() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock);
+
+    TestProtobuf.RepeatedIntMessage.Builder msg = TestProtobuf.RepeatedIntMessage.newBuilder();
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3RepeatedIntMessageSpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.RepeatedIntMessage.class, readConsumerMock, conf);
+
+    TestProto3.RepeatedIntMessage.Builder msg = TestProto3.RepeatedIntMessage.newBuilder();
+    msg.addRepeatedInt(1323);
+    msg.addRepeatedInt(54469);
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("repeatedInt", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).addInteger(1323);
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).addInteger(54469);
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("repeatedInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
   public void testProto3RepeatedIntMessage() throws Exception {
     RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
     ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.RepeatedIntMessage.class, readConsumerMock);
@@ -124,6 +240,268 @@ public class ProtoWriteSupportTest {
   }
 
   @Test
+  public void testProto3RepeatedIntMessageEmptySpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.RepeatedIntMessage.Builder msg = TestProtobuf.RepeatedIntMessage.newBuilder();
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3RepeatedIntMessageEmpty() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.RepeatedIntMessage.class, readConsumerMock);
+
+    TestProtobuf.RepeatedIntMessage.Builder msg = TestProtobuf.RepeatedIntMessage.newBuilder();
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testMapIntMessageSpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.MapIntMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.MapIntMessage.Builder msg = TestProtobuf.MapIntMessage.newBuilder();
+    msg.putMapInt(123, 1);
+    msg.putMapInt(234, 2);
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("mapInt", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key_value", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(123);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(1);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(234);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(2);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("key_value", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("mapInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testMapIntMessage() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.MapIntMessage.class, readConsumerMock);
+
+    TestProtobuf.MapIntMessage.Builder msg = TestProtobuf.MapIntMessage.newBuilder();
+    msg.putMapInt(123, 1);
+    msg.putMapInt(234, 2);
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("mapInt", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(123);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(1);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(234);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(2);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("mapInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testMapIntMessageEmptySpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.MapIntMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.MapIntMessage.Builder msg = TestProtobuf.MapIntMessage.newBuilder();
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testMapIntMessageEmpty() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.MapIntMessage.class, readConsumerMock);
+
+    TestProtobuf.MapIntMessage.Builder msg = TestProtobuf.MapIntMessage.newBuilder();
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3MapIntMessageSpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.MapIntMessage.class, readConsumerMock, conf);
+
+    TestProto3.MapIntMessage.Builder msg = TestProto3.MapIntMessage.newBuilder();
+    msg.putMapInt(123, 1);
+    msg.putMapInt(234, 2);
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("mapInt", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key_value", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(123);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(1);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(234);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(2);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("key_value", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("mapInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3MapIntMessage() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.MapIntMessage.class, readConsumerMock);
+
+    TestProto3.MapIntMessage.Builder msg = TestProto3.MapIntMessage.newBuilder();
+    msg.putMapInt(123, 1);
+    msg.putMapInt(234, 2);
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("mapInt", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(123);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(1);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("key", 0);
+    inOrder.verify(readConsumerMock).addInteger(234);
+    inOrder.verify(readConsumerMock).endField("key", 0);
+    inOrder.verify(readConsumerMock).startField("value", 1);
+    inOrder.verify(readConsumerMock).addInteger(2);
+    inOrder.verify(readConsumerMock).endField("value", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("mapInt", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3MapIntMessageEmptySpecsCompliant() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.MapIntMessage.class, readConsumerMock, conf);
+
+    TestProto3.MapIntMessage.Builder msg = TestProto3.MapIntMessage.newBuilder();
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3MapIntMessageEmpty() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.MapIntMessage.class, readConsumerMock);
+
+    TestProto3.MapIntMessage.Builder msg = TestProto3.MapIntMessage.newBuilder();
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
   public void testRepeatedInnerMessageMessage_message() throws Exception {
     RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
     ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.TopMessage.class, readConsumerMock);
@@ -137,6 +515,7 @@ public class ProtoWriteSupportTest {
 
     inOrder.verify(readConsumerMock).startMessage();
     inOrder.verify(readConsumerMock).startField("inner", 0);
+
     inOrder.verify(readConsumerMock).startGroup();
     inOrder.verify(readConsumerMock).startField("one", 0);
     inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
@@ -145,14 +524,54 @@ public class ProtoWriteSupportTest {
     inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
     inOrder.verify(readConsumerMock).endField("two", 1);
     inOrder.verify(readConsumerMock).endGroup();
+
     inOrder.verify(readConsumerMock).endField("inner", 0);
     inOrder.verify(readConsumerMock).endMessage();
     Mockito.verifyNoMoreInteractions(readConsumerMock);
   }
 
   @Test
-  public void testProto3RepeatedInnerMessageMessage_message() throws Exception {
+  public void testRepeatedInnerMessageSpecsCompliantMessage_message() throws Exception {
     RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.TopMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.TopMessage.Builder msg = TestProtobuf.TopMessage.newBuilder();
+    msg.addInnerBuilder().setOne("one").setTwo("two");
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("inner", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("one", 0);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
+    inOrder.verify(readConsumerMock).endField("one", 0);
+    inOrder.verify(readConsumerMock).startField("two", 1);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
+    inOrder.verify(readConsumerMock).endField("two", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("inner", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3RepeatedInnerMessageMessage_message() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);;
     ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.TopMessage.class, readConsumerMock);
 
     TestProto3.TopMessage.Builder msg = TestProto3.TopMessage.newBuilder();
@@ -164,6 +583,7 @@ public class ProtoWriteSupportTest {
 
     inOrder.verify(readConsumerMock).startMessage();
     inOrder.verify(readConsumerMock).startField("inner", 0);
+
     inOrder.verify(readConsumerMock).startGroup();
     inOrder.verify(readConsumerMock).startField("one", 0);
     inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
@@ -172,6 +592,96 @@ public class ProtoWriteSupportTest {
     inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
     inOrder.verify(readConsumerMock).endField("two", 1);
     inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("inner", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
+  public void testProto3RepeatedInnerMessageSpecsCompliantMessage_message() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.TopMessage.class, readConsumerMock, conf);
+
+    TestProto3.TopMessage.Builder msg = TestProto3.TopMessage.newBuilder();
+    msg.addInnerBuilder().setOne("one").setTwo("two");
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("inner", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("one", 0);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
+    inOrder.verify(readConsumerMock).endField("one", 0);
+    inOrder.verify(readConsumerMock).startField("two", 1);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
+    inOrder.verify(readConsumerMock).endField("two", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("inner", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+
+  @Test
+  public void testRepeatedInnerMessageSpecsCompliantMessage_scalar() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.TopMessage.class, readConsumerMock, conf);
+
+    TestProtobuf.TopMessage.Builder msg = TestProtobuf.TopMessage.newBuilder();
+    msg.addInnerBuilder().setOne("one");
+    msg.addInnerBuilder().setTwo("two");
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("inner", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+
+    //first inner message
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("one", 0);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
+    inOrder.verify(readConsumerMock).endField("one", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    //second inner message
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("two", 1);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
+    inOrder.verify(readConsumerMock).endField("two", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
     inOrder.verify(readConsumerMock).endField("inner", 0);
     inOrder.verify(readConsumerMock).endMessage();
     Mockito.verifyNoMoreInteractions(readConsumerMock);
@@ -192,6 +702,7 @@ public class ProtoWriteSupportTest {
 
     inOrder.verify(readConsumerMock).startMessage();
     inOrder.verify(readConsumerMock).startField("inner", 0);
+
     //first inner message
     inOrder.verify(readConsumerMock).startGroup();
     inOrder.verify(readConsumerMock).startField("one", 0);
@@ -226,6 +737,7 @@ public class ProtoWriteSupportTest {
 
     inOrder.verify(readConsumerMock).startMessage();
     inOrder.verify(readConsumerMock).startField("inner", 0);
+
     //first inner message
     inOrder.verify(readConsumerMock).startGroup();
     inOrder.verify(readConsumerMock).startField("one", 0);
@@ -246,6 +758,55 @@ public class ProtoWriteSupportTest {
   }
 
   @Test
+  public void testProto3RepeatedInnerMessageSpecsCompliantMessage_scalar() throws Exception {
+    RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
+    Configuration conf = new Configuration();
+    ProtoWriteSupport.setWriteSpecsCompliant(conf, true);
+    ProtoWriteSupport instance = createReadConsumerInstance(TestProto3.TopMessage.class, readConsumerMock, conf);
+
+    TestProto3.TopMessage.Builder msg = TestProto3.TopMessage.newBuilder();
+    msg.addInnerBuilder().setOne("one");
+    msg.addInnerBuilder().setTwo("two");
+
+    instance.write(msg.build());
+
+    InOrder inOrder = Mockito.inOrder(readConsumerMock);
+
+    inOrder.verify(readConsumerMock).startMessage();
+    inOrder.verify(readConsumerMock).startField("inner", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("list", 0);
+
+    //first inner message
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("one", 0);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("one".getBytes()));
+    inOrder.verify(readConsumerMock).endField("one", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    //second inner message
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("element", 0);
+    inOrder.verify(readConsumerMock).startGroup();
+    inOrder.verify(readConsumerMock).startField("two", 1);
+    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("two".getBytes()));
+    inOrder.verify(readConsumerMock).endField("two", 1);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("element", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+
+    inOrder.verify(readConsumerMock).endField("list", 0);
+    inOrder.verify(readConsumerMock).endGroup();
+    inOrder.verify(readConsumerMock).endField("inner", 0);
+    inOrder.verify(readConsumerMock).endMessage();
+    Mockito.verifyNoMoreInteractions(readConsumerMock);
+  }
+
+  @Test
   public void testOptionalInnerMessage() throws Exception {
     RecordConsumer readConsumerMock =  Mockito.mock(RecordConsumer.class);
     ProtoWriteSupport instance = createReadConsumerInstance(TestProtobuf.MessageA.class, readConsumerMock);
diff --git a/parquet-protobuf/src/test/java/org/apache/parquet/proto/utils/WriteUsingMR.java b/parquet-protobuf/src/test/java/org/apache/parquet/proto/utils/WriteUsingMR.java
index d18076a..55f9237 100644
--- a/parquet-protobuf/src/test/java/org/apache/parquet/proto/utils/WriteUsingMR.java
+++ b/parquet-protobuf/src/test/java/org/apache/parquet/proto/utils/WriteUsingMR.java
@@ -46,10 +46,18 @@ import static java.lang.Thread.sleep;
 public class WriteUsingMR {
 
   private static final Logger LOG = LoggerFactory.getLogger(WriteUsingMR.class);
-  Configuration conf = new Configuration();
+  private final Configuration conf;
   private static List<Message> inputMessages;
   Path outputPath;
 
+  public WriteUsingMR() {
+    this(new Configuration());
+  }
+
+  public WriteUsingMR(Configuration conf) {
+    this.conf = new Configuration();
+  }
+
   public Configuration getConfiguration() {
     return conf;
   }
diff --git a/parquet-protobuf/src/test/resources/TestProto3.proto b/parquet-protobuf/src/test/resources/TestProto3.proto
index 1896445..e49eef5 100644
--- a/parquet-protobuf/src/test/resources/TestProto3.proto
+++ b/parquet-protobuf/src/test/resources/TestProto3.proto
@@ -124,6 +124,14 @@ message RepeatedIntMessage {
     repeated int32 repeatedInt = 1;
 }
 
+message RepeatedInnerMessage {
+    repeated InnerMessage repeatedInnerMessage = 1;
+}
+
+message MapIntMessage {
+    map<int32, int32> mapInt = 1;
+}
+
 message HighIndexMessage {
     repeated int32 repeatedInt = 50000;
 }
diff --git a/parquet-protobuf/src/test/resources/TestProtobuf.proto b/parquet-protobuf/src/test/resources/TestProtobuf.proto
index d7cdf03..d4ab4c7 100644
--- a/parquet-protobuf/src/test/resources/TestProtobuf.proto
+++ b/parquet-protobuf/src/test/resources/TestProtobuf.proto
@@ -122,6 +122,14 @@ message RepeatedIntMessage {
     repeated int32 repeatedInt = 1;
 }
 
+message RepeatedInnerMessage {
+    repeated InnerMessage repeatedInnerMessage = 1;
+}
+
+message MapIntMessage {
+    map<int32, int32> mapInt = 1;
+}
+
 message HighIndexMessage {
     repeated int32 repeatedInt = 50000;
 }

-- 
To stop receiving notification emails like this one, please contact
julien@apache.org.

Mime
View raw message