initial commit

Author: mihailoale-db

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

@@ -17,29 +17,101 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.catalyst.expressions.{AliasHelper, EvalHelper, Expression}
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions.{AliasHelper, EvalHelper, Expression, SubqueryExpression, VariableReference}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{CreateView, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_IDENTIFIER
+import org.apache.spark.sql.connector.catalog.CatalogV2Util.isSessionCatalog
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.types.StringType
 
 /**
  * Resolves the identifier expressions and builds the original plans/expressions.
  */
 object ResolveIdentifierClause extends Rule[LogicalPlan] with AliasHelper with EvalHelper {
 
-  override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
-    _.containsPattern(UNRESOLVED_IDENTIFIER)) {
+  override def apply(plan: LogicalPlan): LogicalPlan = {
+    val referredTempVars = new mutable.ArrayBuffer[Seq[String]]
+    val analyzedPlan = apply0(plan, referredTempVars)
+
+    analyzedPlan match {
+      case cv @ CreateView(
+      ResolvedIdentifierInSessionCatalog(ident), _, _, _, _, _, _, _, _, _) =>
+        if (referredTempVars.nonEmpty) {
+          throw QueryCompilationErrors.notAllowedToCreatePermanentViewByReferencingTempVarError(
+            ident,
+            referredTempVars.head.quoted
+          )
+        }
+        cv
+      case _ => analyzedPlan
+    }
+  }
+
+  object ResolvedIdentifierInSessionCatalog{
+    def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
+      case ResolvedIdentifier(catalog, ident) if isSessionCatalog(catalog) =>
+        if (ident.namespace().length != 1) {
+          throw QueryCompilationErrors
+            .requiresSinglePartNamespaceError(ident.namespace().toImmutableArraySeq)
+        }
+        Some(TableIdentifier(ident.name, Some(ident.namespace.head), Some(catalog.name)))
+      case _ => None
+    }
+  }
+
+  private def apply0(
+      plan: LogicalPlan,
+      referredTempVars: mutable.ArrayBuffer[Seq[String]]): LogicalPlan =
+    plan.resolveOperatorsUpWithPruning(_.containsPattern(UNRESOLVED_IDENTIFIER)) {
     case p: PlanWithUnresolvedIdentifier if p.identifierExpr.resolved && p.childrenResolved =>
+
+      referredTempVars ++= collectTemporaryVariablesInLogicalPlan(p)
+
       p.planBuilder.apply(evalIdentifierExpr(p.identifierExpr), p.children)
     case other =>
       other.transformExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_IDENTIFIER)) {
         case e: ExpressionWithUnresolvedIdentifier if e.identifierExpr.resolved =>
+
+          referredTempVars ++= collectTemporaryVariablesInExpressionTree(e)
+
           e.exprBuilder.apply(evalIdentifierExpr(e.identifierExpr), e.otherExprs)
       }
   }
 
+  /**
+   * Collect all temporary SQL variables and return the identifiers separately.
+   */
+  private def collectTemporaryVariablesInLogicalPlan(child: LogicalPlan): Seq[Seq[String]] = {
+    def collectTempVars(child: LogicalPlan): Seq[Seq[String]] = {
+      child.flatMap { plan =>
+        plan.expressions.flatMap(_.flatMap {
+          case e: SubqueryExpression => collectTempVars(e.plan)
+          case r: VariableReference => Seq(r.originalNameParts)
+          case _ => Seq.empty
+        })
+      }.distinct
+    }
+    collectTempVars(child)
+  }
+
+  private def collectTemporaryVariablesInExpressionTree(child: Expression): Seq[Seq[String]] = {
+    def collectTempVars(child: Expression): Seq[Seq[String]] = {
+      child.flatMap { expr =>
+        expr.children.flatMap(_.flatMap {
+          case e: SubqueryExpression => collectTemporaryVariablesInLogicalPlan(e.plan)
+          case r: VariableReference => Seq(r.originalNameParts)
+          case _ => Seq.empty
+        })
+      }.distinct
+    }
+    collectTempVars(child)
+  }
+
   private def evalIdentifierExpr(expr: Expression): Seq[String] = {
     trimAliases(prepareForEval(expr)) match {
       case e if !e.foldable => expr.failAnalysis(

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewTestSuite.scala

@@ -793,6 +793,23 @@ class PersistedViewTestSuite extends SQLViewTestSuite with SharedSparkSession {
     }
   }
 
+  test("SPARK-51552: Temporary variables under identifiers are not allowed in persisted view") {
+    sql("declare table_name = 'table';")
+    sql("create table identifier(table_name) (c1 int);")
+    checkError(
+      exception = intercept[AnalysisException] {
+        sql("create view identifier('v_' || table_name) as select * from identifier(table_name);")
+      },
+      condition = "INVALID_TEMP_OBJ_REFERENCE",
+      parameters = Map(
+        "obj" -> "VIEW",
+        "objName" -> "`spark_catalog`.`default`.`v_table`",
+        "tempObj" -> "VARIABLE",
+        "tempObjName" -> "`table_name`"
+      )
+    )
+  }
+
   def getShowCreateDDL(view: String, serde: Boolean = false): String = {
     val result = if (serde) {
       sql(s"SHOW CREATE TABLE $view AS SERDE")