address comments

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -437,7 +437,7 @@ class SessionCatalog(
   /**
    * Retrieve all metadata of existing permanent tables/views. If no database is specified,
    * assume the table/view is in the current database.
-   * Only the tables/views belong to one same database that can be retrieved are returned.
+   * Only the tables/views belong to the same database that can be retrieved are returned.
    * For example, if none of the requested tables could be retrieved, an empty list is returned.
    * There is no guarantee of ordering of the returned tables.
    */
@@ -447,10 +447,10 @@ class SessionCatalog(
       val dbs = names.map(_.database.getOrElse(getCurrentDatabase))
       if (dbs.distinct.size != 1) {
         val tables = names.map(name => formatTableName(name.table))
-        dbs.zip(tables).map { case (d, t) => QualifiedTableName(d, t)}
+        val qualifiedTableNames = dbs.zip(tables).map { case (d, t) => QualifiedTableName(d, t)}
         throw new AnalysisException(
-          s"Only the tables/views belong to one same database can be retrieved. Querying " +
-          s"tables/views are ${dbs.zip(tables).map { case (d, t) => QualifiedTableName(d, t)}}"
+          s"Only the tables/views belong to the same database can be retrieved. Querying " +
+          s"tables/views are $qualifiedTableNames"
         )
       }
       val db = formatDatabaseName(dbs.head)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogSuite.scala

@@ -287,6 +287,14 @@ abstract class ExternalCatalogSuite extends SparkFunSuite with BeforeAndAfterEac
       .map(_.identifier.table) == Seq("tbl1"))
   }
 
+  test("get tables by name when contains invalid name") {
+    // scalastyle:off
+    val name = "砖"
+    // scalastyle:on
+    assert(newBasicCatalog().getTablesByName("db2", Seq("tbl1", name))
+      .map(_.identifier.table) == Seq("tbl1"))
+  }
+
   test("get tables by name when empty table list") {
     assert(newBasicCatalog().getTablesByName("db2", Seq.empty).isEmpty)
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -547,6 +547,28 @@ abstract class SessionCatalogSuite extends AnalysisTest {
     }
   }
 
+  test("get tables by name when contains invalid name") {
+    // scalastyle:off
+    val name = "砖"
+    // scalastyle:on
+    withBasicCatalog { catalog =>
+      assert(catalog.getTablesByName(
+        Seq(
+          TableIdentifier("tbl1", Some("db2")),
+          TableIdentifier(name, Some("db2"))
+        )
+      ) == catalog.externalCatalog.getTablesByName("db2", Seq("tbl1")))
+      // Get table without explicitly specifying database
+      catalog.setCurrentDatabase("db2")
+      assert(catalog.getTablesByName(
+        Seq(
+          TableIdentifier("tbl1"),
+          TableIdentifier(name)
+        )
+      ) == catalog.externalCatalog.getTablesByName("db2", Seq("tbl1")))
+    }
+  }
+
   test("get tables by name when empty") {
     withBasicCatalog { catalog =>
       assert(catalog.getTablesByName(Seq.empty)

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala

@@ -148,7 +148,6 @@ class SparkMetadataOperationSuite extends HiveThriftJdbcTest {
 
     withJdbcStatement("table1", "table2") { statement =>
       Seq(
-        "DROP VIEW IF EXISTS view1",
         "CREATE TABLE table1(key INT, val STRING)",
         "CREATE TABLE table2(key INT, val STRING)",
         "CREATE VIEW view1 AS SELECT * FROM table2").foreach(statement.execute)

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala

@@ -84,6 +84,7 @@ private[hive] trait HiveClient {
   /** Returns the metadata for the specified table or None if it doesn't exist. */
   def getTableOption(dbName: String, tableName: String): Option[CatalogTable]
 
+  /** Returns metadata of existing permanent tables/views for given names. */
   def getTablesByName(dbName: String, tableNames: Seq[String]): Seq[CatalogTable]
 
   /** Creates a table with the given metadata. */

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -1124,22 +1124,24 @@ private[hive] object HiveClientImpl {
     // For non-views, we need to do some extra fixes
     if (!(HiveTableType.VIRTUAL_VIEW.toString == tTable.getTableType)) {
       // Fix the non-printable chars
-      val parameters: JMap[String, String] = tTable.getSd.getParameters
-      val sf: String = parameters.get(serdeConstants.SERIALIZATION_FORMAT)
-      if (sf != null) {
-        val b: Array[Char] = sf.toCharArray
-        if ((b.length == 1) && (b(0) < 10)) { // ^A, ^B, ^C, ^D, \t
-          parameters.put(serdeConstants.SERIALIZATION_FORMAT, Integer.toString(b(0)))
+      val parameters = tTable.getSd.getParameters
+      if (parameters != null) {
+        val sf = parameters.get(serdeConstants.SERIALIZATION_FORMAT)
+        if (sf != null) {
+          val b: Array[Char] = sf.toCharArray
+          if ((b.length == 1) && (b(0) < 10)) { // ^A, ^B, ^C, ^D, \t
+            parameters.put(serdeConstants.SERIALIZATION_FORMAT, Integer.toString(b(0)))
+          }
         }
       }
       // Use LazySimpleSerDe for MetadataTypedColumnsetSerDe.
       // NOTE: LazySimpleSerDe does not support tables with a single column of col
       // of type "array<string>". This happens when the table is created using
       // an earlier version of Hive.
-      if (classOf[MetadataTypedColumnsetSerDe].getName
-            == tTable.getSd.getSerdeInfo.getSerializationLib
-            && tTable.getSd.getColsSize > 0
-            && tTable.getSd.getCols.get(0).getType.indexOf('<') == -1) {
+      if (classOf[MetadataTypedColumnsetSerDe].getName ==
+        tTable.getSd.getSerdeInfo.getSerializationLib &&
+        tTable.getSd.getColsSize > 0 &&
+        tTable.getSd.getCols.get(0).getType.indexOf('<') == -1) {
         tTable.getSd.getSerdeInfo.setSerializationLib(classOf[LazySimpleSerDe].getName)
       }
     }

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala

@@ -252,6 +252,14 @@ class VersionsSuite extends SparkFunSuite with Logging {
         .map(_.identifier.table) == Seq("src"))
     }
 
+    test(s"$version: getTablesByName when contains invalid name") {
+      // scalastyle:off
+      val name = "砖"
+      // scalastyle:on
+      assert(client.getTablesByName("default", Seq("src", name))
+        .map(_.identifier.table) == Seq("src"))
+    }
+
     test(s"$version: getTablesByName when empty") {
       assert(client.getTablesByName("default", Seq.empty).isEmpty)
     }