[SPARK-2287] [SQL] Make ScalaReflection be able to handle Generic case classes.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -47,10 +47,13 @@ object ScalaReflection {
       val TypeRef(_, _, Seq(optType)) = t
       Schema(schemaFor(optType).dataType, nullable = true)
     case t if t <:< typeOf[Product] =>
-      val params = t.member("<init>": TermName).asMethod.paramss
+      val formalTypeArgs = t.typeSymbol.asClass.typeParams
+      val TypeRef(_, _, actualTypeArgs) = t
+      val params = t.member(nme.CONSTRUCTOR).asMethod.paramss
       Schema(StructType(
         params.head.map { p =>
-          val Schema(dataType, nullable) = schemaFor(p.typeSignature)
+          val Schema(dataType, nullable) =
+            schemaFor(p.typeSignature.substituteTypes(formalTypeArgs, actualTypeArgs))
           StructField(p.name.toString, dataType, nullable)
         }), nullable = true)
     // Need to decide if we actually need a special type here.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ScalaReflectionSuite.scala

@@ -60,6 +60,9 @@ case class ComplexData(
     mapField: Map[Int, String],
     structField: PrimitiveData)
 
+case class GenericData[A](
+    genericField: A)
+
 class ScalaReflectionSuite extends FunSuite {
   import ScalaReflection._
 
@@ -128,4 +131,21 @@ class ScalaReflectionSuite extends FunSuite {
           nullable = true))),
       nullable = true))
   }
+
+  test("generic data") {
+    val schema = schemaFor[GenericData[Int]]
+    assert(schema === Schema(
+      StructType(Seq(
+        StructField("genericField", IntegerType, nullable = false))),
+      nullable = true))
+  }
+
+  test("tuple data") {
+    val schema = schemaFor[(Int, String)]
+    assert(schema === Schema(
+      StructType(Seq(
+        StructField("_1", IntegerType, nullable = false),
+        StructField("_2", StringType, nullable = true))),
+      nullable = true))
+  }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala

@@ -26,6 +26,9 @@ import scala.collection.JavaConversions._
  * A set of test cases that validate partition and column pruning.
  */
 class PruningSuite extends HiveComparisonTest {
+  // MINOR HACK: You must run a query before calling reset the first time.
+  TestHive.hql("SHOW TABLES")
+
   // Column/partition pruning is not implemented for `InMemoryColumnarTableScan` yet, need to reset
   // the environment to ensure all referenced tables in this suites are not cached in-memory.
   // Refer to https://issues.apache.org/jira/browse/SPARK-2283 for details.