fix bug of scan an empty parquet based table

Author: chenghao-intel

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

@@ -287,7 +287,11 @@ case class ParquetRelation2(
         }
       }
 
-      parquetSchema = maybeSchema.getOrElse(readSchema())
+      try {
+        parquetSchema = readSchema().getOrElse(maybeSchema.get)
+      } catch {
+        case e => throw new SparkException(s"Failed to find schema for ${paths.mkString(",")}", e)
+      }
 
       partitionKeysIncludedInParquetSchema =
         isPartitioned &&
@@ -308,7 +312,7 @@ case class ParquetRelation2(
       }
     }
 
-    private def readSchema(): StructType = {
+    private def readSchema(): Option[StructType] = {
       // Sees which file(s) we need to touch in order to figure out the schema.
       val filesToTouch =
       // Always tries the summary files first if users don't require a merged schema.  In this case,
@@ -611,8 +615,9 @@ object ParquetRelation2 {
   // internally.
   private[sql] val METASTORE_SCHEMA = "metastoreSchema"
 
-  private[parquet] def readSchema(footers: Seq[Footer], sqlContext: SQLContext): StructType = {
-    footers.map { footer =>
+  private[parquet] def readSchema(
+      footers: Seq[Footer], sqlContext: SQLContext): Option[StructType] = {
+    Option(footers.map { footer =>
       val metadata = footer.getParquetMetadata.getFileMetaData
       val parquetSchema = metadata.getSchema
       val maybeSparkSchema = metadata
@@ -630,11 +635,12 @@ object ParquetRelation2 {
             sqlContext.conf.isParquetBinaryAsString,
             sqlContext.conf.isParquetINT96AsTimestamp))
       }
-    }.reduce { (left, right) =>
-      try left.merge(right) catch { case e: Throwable =>
-        throw new SparkException(s"Failed to merge incompatible schemas $left and $right", e)
-      }
-    }
+    }.foldLeft[StructType](null) {
+      case (null, right) => right
+      case (left, right) => try left.merge(right) catch { case e: Throwable =>
+          throw new SparkException(s"Failed to merge incompatible schemas $left and $right", e)
+        }
+    })
   }
 
   /**

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -212,8 +212,7 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
           Some(partitionSpec))(hive))
     } else {
       val paths = Seq(metastoreRelation.hiveQlTable.getDataLocation.toString)
-      LogicalRelation(
-        ParquetRelation2(
+      LogicalRelation(ParquetRelation2(
           paths,
           Map(ParquetRelation2.METASTORE_SCHEMA -> metastoreSchema.json),
           Some(metastoreSchema))(hive))

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

@@ -121,13 +121,50 @@ class ParquetDataSourceOnMetastoreSuite extends ParquetMetastoreSuiteBase {
 
   override def beforeAll(): Unit = {
     super.beforeAll()
+
+    sql(s"""
+      create table test_parquet
+      (
+        intField INT,
+        stringField STRING
+      )
+      ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
+       STORED AS
+       INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
+       OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
+    """)
+
+    val rdd = sparkContext.parallelize((1 to 10).map(i => s"""{"a":$i, "b":"str${i}"}"""))
+    jsonRDD(rdd).registerTempTable("jt")
+    sql("""
+      create table test ROW FORMAT
+          |  SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
+          |  STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
+          |  OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
+          |  AS select * from jt""".stripMargin)
+
     conf.setConf(SQLConf.PARQUET_USE_DATA_SOURCE_API, "true")
   }
 
   override def afterAll(): Unit = {
     super.afterAll()
+    sql("DROP TABLE test_parquet")
+    sql("DROP TABLE test")
+    sql("DROP TABLE jt")
+
     setConf(SQLConf.PARQUET_USE_DATA_SOURCE_API, originalConf.toString)
   }
+
+  test("scan from an empty parquet table") {
+    checkAnswer(sql("SELECT count(*) FROM test_parquet"), Row(0))
+  }
+
+  test("scan from an non empty parquet table") {
+    checkAnswer(
+      sql(s"SELECT a, b FROM jt WHERE a = '1'"),
+      Seq(Row(1, "str1"))
+    )
+  }
 }
 
 class ParquetDataSourceOffMetastoreSuite extends ParquetMetastoreSuiteBase {