Fixing one problem with nested arrays

Author: AndreSchumacher

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala

@@ -206,8 +206,11 @@ private[parquet] object ParquetTypesConverter {
           val fields = groupType.getFields.map {
             field => new StructField(field.getName, toDataType(field), field.getRepetition != Repetition.REQUIRED)
           }
-          if (fields.size == 1) new ArrayType(fields.apply(0).dataType)
-          new ArrayType(StructType(fields))
+          if (fields.size == 1) {
+            new ArrayType(fields.apply(0).dataType)
+          } else {
+            new ArrayType(StructType(fields))
+          }
         }
         case _ => { // everything else nested becomes a Struct, unless it has a single repeated field
           // in which case it becomes an array (this should correspond to the inverse operation of
@@ -260,7 +263,7 @@ private[parquet] object ParquetTypesConverter {
           elementType match {
             case StructType(fields) => { // first case: array of structs
               val parquetFieldTypes = fields.map(f => fromDataType(f.dataType, f.name, f.nullable, false))
-              new ParquetGroupType(Repetition.REPEATED, name, ParquetOriginalType.LIST, parquetFieldTypes)
+              new ParquetGroupType(repetition, name, ParquetOriginalType.LIST, parquetFieldTypes)
               //ConversionPatterns.listType(Repetition.REPEATED, name, parquetFieldTypes)
             }
             case _ => { // second case: array of primitive types

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTableSupport.scala

@@ -157,7 +157,7 @@ private[parquet] object CatalystConverter {
       case ArrayType(elementType: DataType) => {
         elementType match {
           case StructType(fields) =>
-            if (fields.size > 1) new CatalystGroupConverter(fields, fieldIndex, parent) //CatalystStructArrayConverter(fields, fieldIndex, parent)
+            if (fields.size > 1) new CatalystGroupConverter(fields, fieldIndex, parent)
             else new CatalystArrayConverter(fields(0).dataType, fieldIndex, parent)
           case _ => new CatalystArrayConverter(elementType, fieldIndex, parent)
         }
@@ -244,8 +244,11 @@ class CatalystGroupConverter(
   override val size = schema.size
 
   // Should be only called in root group converter!
-  def getCurrentRecord: Row = new GenericRow {
-    override val values: Array[Any] = current.toArray
+  def getCurrentRecord: Row = {
+    assert(isRootConverter, "getCurrentRecord should only be called in root group converter!")
+    new GenericRow {
+      override val values: Array[Any] = current.toArray
+    }
   }
 
   override def getConverter(fieldIndex: Int): Converter = converters(fieldIndex)

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTestData.scala

@@ -143,10 +143,10 @@ private[sql] object ParquetTestData {
           |required double value;
           |optional boolean truth;
         |}
-        |required group outerouter {
-          |required group outer {
-            |required group inner {
-              |required int32 number;
+        |optional group outerouter {
+          |repeated group values {
+            |repeated group values {
+              |repeated int32 values;
             |}
           |}
         |}
@@ -263,9 +263,9 @@ private[sql] object ParquetTestData {
     val booleanNumberPairs = r1.addGroup(3)
     booleanNumberPairs.add("value", 2.5)
     booleanNumberPairs.add("truth", false)
-    r1.addGroup(4).addGroup(0).addGroup(0).add("number", 7)
-    r1.addGroup(4).addGroup(0).addGroup(0).add("number", 8)
-    r1.addGroup(4).addGroup(0).addGroup(0).add("number", 9)
+    r1.addGroup(4).addGroup(0).addGroup(0).add("values", 7)
+    r1.addGroup(4).addGroup(0).addGroup(0).add("values", 8)
+    r1.addGroup(4).addGroup(0).addGroup(0).add("values", 9)
 
     val writeSupport = new TestGroupWriteSupport(schema)
     val writer = new ParquetWriter[Group](path, writeSupport)