[SPARK-17860][SQL] SHOW COLUMN's database conflict check should respect case sensitivity configuration

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -168,17 +168,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
    * }}}
    */
   override def visitShowColumns(ctx: ShowColumnsContext): LogicalPlan = withOrigin(ctx) {
-    val table = visitTableIdentifier(ctx.tableIdentifier)
-
-    val lookupTable = Option(ctx.db) match {
-      case None => table
-      case Some(db) if table.database.exists(_ != db) =>
-        operationNotAllowed(
-          s"SHOW COLUMNS with conflicting databases: '$db' != '${table.database.get}'",
-          ctx)
-      case Some(db) => TableIdentifier(table.identifier, Some(db.getText))
-    }
-    ShowColumnsCommand(lookupTable)
+    ShowColumnsCommand(Option(ctx.db).map(_.getText), visitTableIdentifier(ctx.tableIdentifier))
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -651,14 +651,24 @@ case class ShowTablePropertiesCommand(table: TableIdentifier, propertyKey: Optio
  *   SHOW COLUMNS (FROM | IN) table_identifier [(FROM | IN) database];
  * }}}
  */
-case class ShowColumnsCommand(tableName: TableIdentifier) extends RunnableCommand {
+case class ShowColumnsCommand(
+    databaseName: Option[String],
+    tableName: TableIdentifier) extends RunnableCommand {
   override val output: Seq[Attribute] = {
     AttributeReference("col_name", StringType, nullable = false)() :: Nil
   }
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getTempViewOrPermanentTableMetadata(tableName)
+    val resolver = sparkSession.sessionState.conf.resolver
+    val lookupTable = databaseName match {
+      case None => tableName
+      case Some(db) if tableName.database.exists(!resolver(_, db)) =>
+        throw new AnalysisException(
+          s"SHOW COLUMNS with conflicting databases: '$db' != '${tableName.database.get}'")
+      case Some(db) => TableIdentifier(tableName.identifier, Some(db))
+    }
+    val table = catalog.getTempViewOrPermanentTableMetadata(lookupTable)
     table.schema.map { c =>
       Row(c.name)
     }

sql/core/src/test/resources/sql-tests/inputs/show_columns.sql

@@ -0,0 +1,58 @@
+CREATE DATABASE showdb;
+
+USE showdb;
+
+CREATE TABLE showcolumn1 (col1 int, `col 2` int);
+CREATE TABLE showcolumn2 (price int, qty int) partitioned by (year int, month int);
+CREATE TEMPORARY VIEW showColumn3 (col3 int, `col 4` int) USING parquet;
+CREATE GLOBAL TEMP VIEW showColumn4 AS SELECT 1 as col1, 'abc' as `col 5`;
+
+
+-- only table name
+SHOW COLUMNS IN showcolumn1;
+
+-- qualified table name
+SHOW COLUMNS IN showdb.showcolumn1;
+
+-- table name and database name
+SHOW COLUMNS IN showcolumn1 FROM showdb;
+
+-- partitioned table
+SHOW COLUMNS IN showcolumn2 IN showdb;
+
+-- Non-existent table. Raise an error in this case
+SHOW COLUMNS IN badtable FROM showdb;
+
+-- database in table identifier and database name in different case
+SHOW COLUMNS IN showdb.showcolumn1 from SHOWDB;
+
+-- different database name in table identifier and database name.
+-- Raise an error in this case.
+SHOW COLUMNS IN showdb.showcolumn1 FROM baddb;
+
+-- show column on temporary view
+SHOW COLUMNS IN showcolumn3;
+
+-- error temp view can't be qualified with a database
+SHOW COLUMNS IN showdb.showcolumn3;
+
+-- error temp view can't be qualified with a database
+SHOW COLUMNS IN showcolumn3 FROM showdb;
+
+-- error global temp view needs to be qualified
+SHOW COLUMNS IN showcolumn4;
+
+-- global temp view qualified with database
+SHOW COLUMNS IN global_temp.showcolumn4;
+
+-- global temp view qualified with database
+SHOW COLUMNS IN showcolumn4 FROM global_temp;
+
+DROP TABLE showcolumn1;
+DROP TABLE showColumn2;
+DROP VIEW  showcolumn3;
+DROP VIEW  global_temp.showcolumn4;
+
+use default;
+
+DROP DATABASE showdb;

sql/core/src/test/resources/sql-tests/results/show_columns.sql.out

@@ -0,0 +1,217 @@
+-- Automatically generated by SQLQueryTestSuite
+-- Number of queries: 25
+
+
+-- !query 0
+CREATE DATABASE showdb
+-- !query 0 schema
+struct<>
+-- !query 0 output
+
+
+
+-- !query 1
+USE showdb
+-- !query 1 schema
+struct<>
+-- !query 1 output
+
+
+
+-- !query 2
+CREATE TABLE showcolumn1 (col1 int, `col 2` int)
+-- !query 2 schema
+struct<>
+-- !query 2 output
+
+
+
+-- !query 3
+CREATE TABLE showcolumn2 (price int, qty int) partitioned by (year int, month int)
+-- !query 3 schema
+struct<>
+-- !query 3 output
+
+
+
+-- !query 4
+CREATE TEMPORARY VIEW showColumn3 (col3 int, `col 4` int) USING parquet
+-- !query 4 schema
+struct<>
+-- !query 4 output
+
+
+
+-- !query 5
+CREATE GLOBAL TEMP VIEW showColumn4 AS SELECT 1 as col1, 'abc' as `col 5`
+-- !query 5 schema
+struct<>
+-- !query 5 output
+
+
+
+-- !query 6
+SHOW COLUMNS IN showcolumn1
+-- !query 6 schema
+struct<col_name:string>
+-- !query 6 output
+col 2
+col1
+
+
+-- !query 7
+SHOW COLUMNS IN showdb.showcolumn1
+-- !query 7 schema
+struct<col_name:string>
+-- !query 7 output
+col 2
+col1
+
+
+-- !query 8
+SHOW COLUMNS IN showcolumn1 FROM showdb
+-- !query 8 schema
+struct<col_name:string>
+-- !query 8 output
+col 2
+col1
+
+
+-- !query 9
+SHOW COLUMNS IN showcolumn2 IN showdb
+-- !query 9 schema
+struct<col_name:string>
+-- !query 9 output
+month
+price
+qty
+year
+
+
+-- !query 10
+SHOW COLUMNS IN badtable FROM showdb
+-- !query 10 schema
+struct<>
+-- !query 10 output
+org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+Table or view 'badtable' not found in database 'showdb';
+
+
+-- !query 11
+SHOW COLUMNS IN showdb.showcolumn1 from SHOWDB
+-- !query 11 schema
+struct<col_name:string>
+-- !query 11 output
+col 2
+col1
+
+
+-- !query 12
+SHOW COLUMNS IN showdb.showcolumn1 FROM baddb
+-- !query 12 schema
+struct<>
+-- !query 12 output
+org.apache.spark.sql.AnalysisException
+SHOW COLUMNS with conflicting databases: 'baddb' != 'showdb';
+
+
+-- !query 13
+SHOW COLUMNS IN showcolumn3
+-- !query 13 schema
+struct<col_name:string>
+-- !query 13 output
+col 4
+col3
+
+
+-- !query 14
+SHOW COLUMNS IN showdb.showcolumn3
+-- !query 14 schema
+struct<>
+-- !query 14 output
+org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+Table or view 'showcolumn3' not found in database 'showdb';
+
+
+-- !query 15
+SHOW COLUMNS IN showcolumn3 FROM showdb
+-- !query 15 schema
+struct<>
+-- !query 15 output
+org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+Table or view 'showcolumn3' not found in database 'showdb';
+
+
+-- !query 16
+SHOW COLUMNS IN showcolumn4
+-- !query 16 schema
+struct<>
+-- !query 16 output
+org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+Table or view 'showcolumn4' not found in database 'showdb';
+
+
+-- !query 17
+SHOW COLUMNS IN global_temp.showcolumn4
+-- !query 17 schema
+struct<col_name:string>
+-- !query 17 output
+col 5
+col1
+
+
+-- !query 18
+SHOW COLUMNS IN showcolumn4 FROM global_temp
+-- !query 18 schema
+struct<col_name:string>
+-- !query 18 output
+col 5
+col1
+
+
+-- !query 19
+DROP TABLE showcolumn1
+-- !query 19 schema
+struct<>
+-- !query 19 output
+
+
+
+-- !query 20
+DROP TABLE showColumn2
+-- !query 20 schema
+struct<>
+-- !query 20 output
+
+
+
+-- !query 21
+DROP VIEW  showcolumn3
+-- !query 21 schema
+struct<>
+-- !query 21 output
+
+
+
+-- !query 22
+DROP VIEW  global_temp.showcolumn4
+-- !query 22 schema
+struct<>
+-- !query 22 output
+
+
+
+-- !query 23
+use default
+-- !query 23 schema
+struct<>
+-- !query 23 output
+
+
+
+-- !query 24
+DROP DATABASE showdb
+-- !query 24 schema
+struct<>
+-- !query 24 output
+

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

@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.catalyst.util.{fileToString, stringToFile}
+import org.apache.spark.sql.execution.command.ShowColumnsCommand
 import org.apache.spark.sql.test.SharedSQLContext
 import org.apache.spark.sql.types.StructType
 

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala

@@ -822,22 +822,24 @@ class DDLCommandSuite extends PlanTest {
     val sql1 = "SHOW COLUMNS FROM t1"
     val sql2 = "SHOW COLUMNS IN db1.t1"
     val sql3 = "SHOW COLUMNS FROM t1 IN db1"
-    val sql4 = "SHOW COLUMNS FROM db1.t1 IN db1"
-    val sql5 = "SHOW COLUMNS FROM db1.t1 IN db2"
+    val sql4 = "SHOW COLUMNS FROM db1.t1 IN db2"
 
     val parsed1 = parser.parsePlan(sql1)
-    val expected1 = ShowColumnsCommand(TableIdentifier("t1", None))
+    val expected1 = ShowColumnsCommand(None, TableIdentifier("t1", None))
     val parsed2 = parser.parsePlan(sql2)
-    val expected2 = ShowColumnsCommand(TableIdentifier("t1", Some("db1")))
+    val expected2 = ShowColumnsCommand(None, TableIdentifier("t1", Some("db1")))
     val parsed3 = parser.parsePlan(sql3)
-    val parsed4 = parser.parsePlan(sql3)
+    val expected3 = ShowColumnsCommand(Some("db1"), TableIdentifier("t1", None))
+    val parsed4 = parser.parsePlan(sql4)
+    val expected4 = ShowColumnsCommand(Some("db2"), TableIdentifier("t1", Some("db1")))
+
     comparePlans(parsed1, expected1)
     comparePlans(parsed2, expected2)
-    comparePlans(parsed3, expected2)
-    comparePlans(parsed4, expected2)
-    assertUnsupported(sql5)
+    comparePlans(parsed3, expected3)
+    comparePlans(parsed4, expected4)
   }
 
+
   test("show partitions") {
     val sql1 = "SHOW PARTITIONS t1"
     val sql2 = "SHOW PARTITIONS db1.t1"

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -1713,4 +1713,21 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
       assert(sql("show user functions").count() === 1L)
     }
   }
+
+  test("show columns - negative test") {
+    // When case sensitivity is true, the user supplied database name in table identifier
+    // should match the supplied database name in case sensitive way.
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      withTempDatabase { db =>
+        val tabName = s"$db.showcolumn"
+        withTable(tabName) {
+          sql(s"CREATE TABLE $tabName(col1 int, col2 string) USING parquet ")
+          val message = intercept[AnalysisException] {
+            sql(s"SHOW COLUMNS IN $db.showcolumn FROM ${db.toUpperCase}")
+          }.getMessage
+          assert(message.contains("SHOW COLUMNS with conflicting databases"))
+        }
+      }
+    }
+  }
 }