Replace special characters '(' and ')' of Parquet schema.

Author: viirya

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

@@ -390,10 +390,11 @@ private[parquet] object ParquetTypesConverter extends Logging {
 
   def convertFromAttributes(attributes: Seq[Attribute],
                             toThriftSchemaNames: Boolean = false): MessageType = {
-    val fields = attributes.map(
-      attribute =>
+    val fields = attributes.map { old_attribute =>
+        val attribute = old_attribute.withName(old_attribute.name.replaceAll("\\((.*)\\)", "[$1]"))
         fromDataType(attribute.dataType, attribute.name, attribute.nullable,
-                     toThriftSchemaNames = toThriftSchemaNames))
+                     toThriftSchemaNames = toThriftSchemaNames)
+    }
     new MessageType("root", fields)
   }
 
@@ -405,7 +406,10 @@ private[parquet] object ParquetTypesConverter extends Logging {
   }
 
   def convertToString(schema: Seq[Attribute]): String = {
-    StructType.fromAttributes(schema).json
+    val replaced_schema = schema.map { attribute =>
+         attribute.withName(attribute.name.replaceAll("\\((.*)\\)", "[$1]"))
+    }
+    StructType.fromAttributes(replaced_schema).json
   }
 
   def writeMetaData(attributes: Seq[Attribute], origPath: Path, conf: Configuration): Unit = {

sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala

@@ -578,6 +578,18 @@ class ParquetDataSourceOnSourceSuite extends ParquetSourceSuiteBase {
 
     sql("DROP TABLE alwaysNullable")
   }
+
+  test("Aggregation attribute names including special chars '(' and ')' should be replaced") {
+    val tempDir = Utils.createTempDir()
+    val filePath = new File(tempDir, "testParquet").getCanonicalPath
+
+    val df = Seq(1,2,3).map(i => (i, i.toString)).toDF("int", "str")
+    val df2 = df.as('x).join(df.as('y), $"x.str" === $"y.str").groupBy("y.str").max("y.int")
+    df2.saveAsParquetFile(filePath)
+    val df3 = parquetFile(filePath)
+    checkAnswer(df3, Row("1", 1) :: Row("2", 2) :: Row("3", 3) :: Nil)
+    assert(df3.columns === Array("str", "MAX[int]"))
+  }
 }
 
 class ParquetDataSourceOffSourceSuite extends ParquetSourceSuiteBase {