[SPARK-1354][SQL] Add tableName as a qualifier for SimpleCatelogy

Fix attribute unresolved when query with table name as a qualifier in SQLContext with SimplCatelog, details please see [SPARK-1354](https://issues.apache.org/jira/browse/SPARK-1354?jql=project%20%3D%20SPARK).

Author: jerryshao <saisai.shao@intel.com>

Closes #272 from jerryshao/qualifier-fix and squashes the following commits:

7950170 [jerryshao] Add tableName as a qualifier for SimpleCatelogy

2 files changed, 29 insertions, 1 deletions

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala
index ff66177a03..e09182dd8d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala
@@ -47,10 +47,11 @@ class SimpleCatalog extends Catalog {
       tableName: String,
       alias: Option[String] = None): LogicalPlan = {
     val table = tables.get(tableName).getOrElse(sys.error(s"Table Not Found: $tableName"))
+    val tableWithQualifiers = Subquery(tableName, table)
 
     // If an alias was specified by the lookup, wrap the plan in a subquery so that attributes are
     // properly qualified with this alias.
-    alias.map(a => Subquery(a.toLowerCase, table)).getOrElse(table)
+    alias.map(a => Subquery(a.toLowerCase, tableWithQualifiers)).getOrElse(tableWithQualifiers)
   }
 }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index fa4a1d5189..4c4fd6dbbe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -216,4 +216,31 @@ class SQLQuerySuite extends QueryTest {
       (null, null, 5, "E") ::
       (null, null, 6, "F") :: Nil)
   }
+
+  test("select with table name as qualifier") {
+    checkAnswer(
+      sql("SELECT testData.value FROM testData WHERE testData.key = 1"),
+      Seq(Seq("1")))
+  }
+
+  test("inner join ON with table name as qualifier") {
+    checkAnswer(
+      sql("SELECT * FROM upperCaseData JOIN lowerCaseData ON lowerCaseData.n = upperCaseData.N"),
+      Seq(
+        (1, "A", 1, "a"),
+        (2, "B", 2, "b"),
+        (3, "C", 3, "c"),
+        (4, "D", 4, "d")))
+  }
+
+  test("qualified select with inner join ON with table name as qualifier") {
+    checkAnswer(
+      sql("SELECT upperCaseData.N, upperCaseData.L FROM upperCaseData JOIN lowerCaseData " +
+        "ON lowerCaseData.n = upperCaseData.N"),
+      Seq(
+        (1, "A"),
+        (2, "B"),
+        (3, "C"),
+        (4, "D")))
+  }
 }