[SPARK-34141][SQL] Remove side effect from ExtractGenerator

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

@@ -2659,32 +2659,31 @@ class Analyzer(override val catalogManager: CatalogManager)
         p
 
       case p @ Project(projectList, child) =>
-        // Holds the resolved generator, if one exists in the project list.
-        var resolvedGenerator: Generate = null
-
-        val newProjectList = projectList
+        val (resolvedGenerator, newProjectList) = projectList
           .map(trimNonTopLevelAliases)
-          .flatMap {
-            case AliasedGenerator(generator, names, outer) if generator.childrenResolved =>
-              // It's a sanity check, this should not happen as the previous case will throw
-              // exception earlier.
-              assert(resolvedGenerator == null, "More than one generator found in SELECT.")
-
-              resolvedGenerator =
-                Generate(
+          .foldLeft((None: Option[Generate], Nil: Seq[NamedExpression])) { (res, e) =>
+            e match {
+              case AliasedGenerator(generator, names, outer) if generator.childrenResolved =>
+                // It's a sanity check, this should not happen as the previous case will throw
+                // exception earlier.
+                assert(res._1.isEmpty, "More than one generator found in SELECT.")
+
+                val g = Generate(
                   generator,
                   unrequiredChildIndex = Nil,
                   outer = outer,
                   qualifier = None,
                   generatorOutput = ResolveGenerate.makeGeneratorOutput(generator, names),
                   child)
 
-              resolvedGenerator.generatorOutput
-            case other => other :: Nil
+                (Some(g), res._2 ++ g.generatorOutput)
+              case other =>
+                (res._1, res._2 :+ other)
+            }
           }
 
-        if (resolvedGenerator != null) {
-          Project(newProjectList, resolvedGenerator)
+        if (resolvedGenerator.isDefined) {
+          Project(newProjectList, resolvedGenerator.get)
         } else {
           p
         }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala

@@ -1018,4 +1018,17 @@ class AnalysisSuite extends AnalysisTest with Matchers {
       )
     }
   }
+
+  test("SPARK-34141: ExtractGenerator with lazy project list") {
+    val b = AttributeReference("b", ArrayType(StringType))()
+
+    val columns = AttributeReference("a", StringType)() :: b :: Nil
+    val explode = Alias(Explode(b), "c")()
+
+    // view is a lazy seq
+    val rel = LocalRelation(output = columns.view)
+    val plan = Project(rel.output ++ (explode :: Nil), rel)
+
+    assertAnalysisSuccess(plan)
+  }
 }