[SPARK-34701][SQL] Introduce AnalysisOnlyCommand that allows its children to be removed once the command is marked as analyzed.

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

@@ -304,7 +304,9 @@ class Analyzer(override val catalogManager: CatalogManager)
     Batch("Subquery", Once,
       UpdateOuterReferences),
     Batch("Cleanup", fixedPoint,
-      CleanupAliases)
+      CleanupAliases),
+    Batch("TransformAfterAnalysis", Once,
+      TransformAfterAnalysis)
   )
 
   /**
@@ -3701,6 +3703,16 @@ object CleanupAliases extends Rule[LogicalPlan] with AliasHelper {
   }
 }
 
+/**
+ * A rule that calls `TransformationAfterAnalysis.transform` for a plan that needs to be transformed
+ * after all the analysis rules are run.
+ */
+object TransformAfterAnalysis extends Rule[LogicalPlan] {
+  override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+    case t: TransformationAfterAnalysis => t.transform
+  }
+}
+
 /**
  * Ignore event time watermark in batch query, which is only supported in Structured Streaming.
  * TODO: add this rule into analyzer rule list.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/TransformationAfterAnalysis.scala

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.plans.logical
+
+/**
+ * A trait that exposes 'transform' which will be called by the `TransformAfterAnalysis` rule
+ * after all other analysis rules are called. One scenario to use this transformation is to
+ * remove any children of a logical plan so that they are not optimized; this is useful for
+ * commands that create a view or a cache because they need to work with analyzed plans.
+ */
+trait TransformationAfterAnalysis {
+  def transform: LogicalPlan
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -866,7 +866,14 @@ case class CacheTableAsSelect(
     plan: LogicalPlan,
     originalText: String,
     isLazy: Boolean,
-    options: Map[String, String]) extends LeafCommand
+    options: Map[String, String],
+    isPlanAnalyzed: Boolean = false) extends LeafCommand with TransformationAfterAnalysis {
+  // `plan` needs to be analyzed, but shouldn't be optimized. Thus, remove `plan` from
+  // children once the analysis phase is finished.
+  override def children: Seq[LogicalPlan] = if (!isPlanAnalyzed) plan :: Nil else Nil
+
+  override def transform: LogicalPlan = copy(isPlanAnalyzed = true)
+}
 
 /**
  * The logical plan of the UNCACHE TABLE command.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala

@@ -830,7 +830,7 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
 }
 
 trait LeafLike[T <: TreeNode[T]] { self: TreeNode[T] =>
-  override final def children: Seq[T] = Nil
+  override def children: Seq[T] = Nil
 }
 
 trait UnaryLike[T <: TreeNode[T]] { self: TreeNode[T] =>

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -3370,10 +3370,11 @@ class Dataset[T] private[sql](
       comment = None,
       properties = Map.empty,
       originalText = None,
-      child = logicalPlan,
+      plan = logicalPlan,
       allowExisting = false,
       replace = replace,
-      viewType = viewType)
+      viewType = viewType,
+      isPlanAnalyzed = true)
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -482,7 +482,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
 
     case CreateViewStatement(
       tbl, userSpecifiedColumns, comment, properties,
-      originalText, child, allowExisting, replace, viewType) if child.resolved =>
+      originalText, child, allowExisting, replace, viewType) =>
 
       val v1TableName = if (viewType != PersistedView) {
         // temp view doesn't belong to any catalog and we shouldn't resolve catalog in the name.
@@ -491,15 +491,16 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
         parseV1Table(tbl, "CREATE VIEW")
       }
       CreateViewCommand(
-        v1TableName.asTableIdentifier,
-        userSpecifiedColumns,
-        comment,
-        properties,
-        originalText,
-        child,
-        allowExisting,
-        replace,
-        viewType)
+        name = v1TableName.asTableIdentifier,
+        userSpecifiedColumns = userSpecifiedColumns,
+        comment = comment,
+        properties = properties,
+        originalText = originalText,
+        plan = child,
+        allowExisting = allowExisting,
+        replace = replace,
+        viewType = viewType,
+        isPlanAnalyzed = false)
 
     case ShowViews(resolved: ResolvedNamespace, pattern, output) =>
       resolved match {

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.analysis.{GlobalTempView, LocalTempView, Pe
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, SessionCatalog, TemporaryViewRelation}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, SubqueryExpression, UserDefinedExpression}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, View}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, TransformationAfterAnalysis, View}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.NamespaceHelper
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -62,15 +62,22 @@ case class CreateViewCommand(
     comment: Option[String],
     properties: Map[String, String],
     originalText: Option[String],
-    child: LogicalPlan,
+    plan: LogicalPlan,
     allowExisting: Boolean,
     replace: Boolean,
-    viewType: ViewType)
-  extends RunnableCommand {
+    viewType: ViewType,
+    isPlanAnalyzed: Boolean)
+  extends RunnableCommand with TransformationAfterAnalysis {
 
   import ViewHelper._
 
-  override def innerChildren: Seq[QueryPlan[_]] = Seq(child)
+  override def innerChildren: Seq[QueryPlan[_]] = Seq(plan)
+
+  // `plan` needs to be analyzed, but shouldn't be optimized. Thus, remove `plan` from
+  // children once the analysis phase is finished.
+  override def children: Seq[LogicalPlan] = if (!isPlanAnalyzed) plan :: Nil else Nil
+
+  override def transform: LogicalPlan = copy(isPlanAnalyzed = true)
 
   if (viewType == PersistedView) {
     require(originalText.isDefined, "'originalText' must be provided to create permanent view")
@@ -96,10 +103,8 @@ case class CreateViewCommand(
   }
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
-    // If the plan cannot be analyzed, throw an exception and don't proceed.
-    val qe = sparkSession.sessionState.executePlan(child)
-    qe.assertAnalyzed()
-    val analyzedPlan = qe.analyzed
+    assert(isPlanAnalyzed)
+    val analyzedPlan = plan
 
     if (userSpecifiedColumns.nonEmpty &&
         userSpecifiedColumns.length != analyzedPlan.output.length) {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CacheTableExec.scala

@@ -101,10 +101,11 @@ case class CacheTableAsSelectExec(
         comment = None,
         properties = Map.empty,
         originalText = Some(originalText),
-        child = query,
+        plan = query,
         allowExisting = false,
         replace = false,
-        viewType = LocalTempView
+        viewType = LocalTempView,
+        isPlanAnalyzed = true
       )
     )
     dataFrameForCachedPlan.logicalPlan