Various fixes.

Author: rxin

sql/catalyst/src/main/scala/org/apache/spark/sql/types/dataTypes.scala

@@ -822,7 +822,8 @@ case class StructType(fields: Array[StructField]) extends DataType with Seq[Stru
    * have a name matching the given name, `null` will be returned.
    */
   def apply(name: String): StructField = {
-    nameToField.getOrElse(name, throw new IllegalArgumentException(s"Field $name does not exist."))
+    nameToField.getOrElse(name,
+      throw new IllegalArgumentException(s"""Field "$name" does not exist."""))
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala

@@ -31,7 +31,7 @@ import org.apache.spark.sql.types.StructType
 
 private[sql] object DataFrame {
   def apply(sqlContext: SQLContext, logicalPlan: LogicalPlan): DataFrame = {
-    new DataFrameImpl(sqlContext, sqlContext.executePlan(logicalPlan))
+    new DataFrameImpl(sqlContext, logicalPlan)
   }
 }
 

sql/core/src/main/scala/org/apache/spark/sql/DataFrameImpl.scala

@@ -260,7 +260,7 @@ private[sql] class DataFrameImpl protected[sql](
 
   override def take(n: Int): Array[Row] = head(n)
 
-  override def collect(): Array[Row] = rdd.collect()
+  override def collect(): Array[Row] = queryExecution.executedPlan.executeCollect()
 
   override def collectAsList(): java.util.List[Row] = java.util.Arrays.asList(rdd.collect() :_*)
 

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

@@ -36,11 +36,9 @@ class DataFrameSuite extends QueryTest {
     intercept[Exception] {
       testData.groupBy("nonExistentName").agg(Map("key" -> "sum"))
     }
-
-    // Uncomment the following once we report the errors properly.
-//    intercept[Exception] {
-//      testData.groupBy("nonExistentName").agg(Map("key" -> "sum"))
-//    }
+    intercept[Exception] {
+      testData.groupBy($"abcd").agg(Map("key" -> "sum"))
+    }
   }
 
   test("table scan") {