resolve names for generator in projection

Author: chenghao-intel

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -576,6 +576,12 @@ class Analyzer(
     /** Extracts a [[Generator]] expression and any names assigned by aliases to their output. */
     private object AliasedGenerator {
       def unapply(e: Expression): Option[(Generator, Seq[String])] = e match {
+        case Alias(g: Generator, name) if g.elementTypes.size > 1 => {
+          // In project, we probably give a default name which is not correct
+          // e.g. SELECT explode(map(key, value)) FROM src;
+          // Let's ignore the specified names for generator in project list
+          Some((g, Nil))
+        }
         case Alias(g: Generator, name) => Some((g, name :: Nil))
         case MultiAlias(g: Generator, names) => Some(g, names)
         case _ => None

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

@@ -555,6 +555,13 @@ class SQLQuerySuite extends QueryTest {
     val col = df("val")
   }
 
+  test("resolve udtf without alias in projection") {
+    val rdd = sparkContext.makeRDD((1 to 2).map(i => s"""{"a":[$i, ${i + 1}]}"""))
+    jsonRDD(rdd).registerTempTable("data")
+    checkAnswer(sql("SELECT explode(map(1, 1)) FROM data LIMIT 1"), Row(1, 1) :: Nil)
+    checkAnswer(sql("SELECT explode(array(1, 1)) FROM data LIMIT 1"), Row(1) :: Nil)
+  }
+
   test("logical.Project should not be resolved if it contains aggregates or generators") {
     // This test is used to test the fix of SPARK-5875.
     // The original issue was that Project's resolved will be true when it contains