[SPARK-49349][SQL] Improve error message for LCA with Generate

common/utils/src/main/resources/error/error-conditions.json

@@ -5129,6 +5129,11 @@
           "Referencing lateral column alias <lca> in the aggregate query both with window expressions and with having clause. Please rewrite the aggregate query by removing the having clause or removing lateral alias reference in the SELECT list."
         ]
       },
+      "LATERAL_COLUMN_ALIAS_IN_GENERATOR" : {
+        "message" : [
+          "Referencing a lateral column alias <lca> in generator expression <generatorExpr>."
+        ]
+      },
       "LATERAL_COLUMN_ALIAS_IN_GROUP_BY" : {
         "message" : [
           "Referencing a lateral column alias via GROUP BY alias/ALL is not supported yet."

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

@@ -340,6 +340,21 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog with QueryErrorsB
           // surrounded with single quotes, or there is a typo in the attribute name.
           case GetMapValue(map, key: Attribute) if isMapWithStringKey(map) && !key.resolved =>
             failUnresolvedAttribute(operator, key, "UNRESOLVED_MAP_KEY")
+
+          case e: Expression if e.containsPattern(LATERAL_COLUMN_ALIAS_REFERENCE) &&
+            operator.expressions.exists {
+              case a: Alias
+                  if e.collect { case l: LateralColumnAliasReference => l.nameParts.head }
+                    .contains(a.name) =>
+                a.exists(_.isInstanceOf[Generator])
+              case _ => false
+            } =>
+            val lcaRefNames =
+              e.collect { case lcaRef: LateralColumnAliasReference => lcaRef.name }.distinct
+            failAnalysis(
+              errorClass = "UNSUPPORTED_FEATURE.LATERAL_COLUMN_ALIAS_IN_GENERATOR",
+              messageParameters =
+                Map("lca" -> toSQLId(lcaRefNames), "generatorExpr" -> toSQLExpr(e)))
         }
 
         // Fail if we still have an unresolved all in group by. This needs to run before the

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

@@ -1365,4 +1365,41 @@ class LateralColumnAliasSuite extends LateralColumnAliasSuiteBase {
     // the states are cleared - a subsequent correct query should succeed
     sql("select 1 as a, a").queryExecution.assertAnalyzed()
   }
+
+  test("SPARK-49349: Improve error message for LCA with Generate") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        sql(
+          s"""
+            |SELECT
+            |  explode(split(name , ',')) AS new_name,
+            |  new_name like 'a%'
+            |FROM $testTable
+            |""".stripMargin)
+      },
+      condition = "UNSUPPORTED_FEATURE.LATERAL_COLUMN_ALIAS_IN_GENERATOR",
+      sqlState = "0A000",
+      parameters = Map(
+        "lca" -> "`new_name`",
+        "generatorExpr" -> "\"unresolvedalias(lateralAliasReference(new_name) LIKE a%)\""))
+
+    checkError(
+      exception = intercept[AnalysisException] {
+        sql(
+          s"""
+             |SELECT
+             |  explode_outer(from_json(name,'array<struct<values:string>>')) as newName,
+             |  size(from_json(newName.values,'array<string>')) +
+             |    size(array(from_json(newName.values,'map<string,string>'))) as size
+             |FROM $testTable
+             |""".stripMargin)
+      },
+      condition = "UNSUPPORTED_FEATURE.LATERAL_COLUMN_ALIAS_IN_GENERATOR",
+      sqlState = "0A000",
+      parameters = Map(
+        "lca" -> "`newName.values`",
+        "generatorExpr" -> ("\"(size(from_json(lateralAliasReference(newName.values), " +
+          "array<string>)) + size(array(from_json(lateralAliasReference(newName.values), " +
+          "map<string,string>)))) AS size\"")))
+  }
 }