[CARMEL-6152] [Follow-up] Calculate more accurate information in the view (#1055)

Author: fenzhu

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

@@ -24,11 +24,12 @@ import org.apache.spark.SparkException
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.CatalogTableType
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, DynamicPruningExpression, Expression, SubqueryExpression}
-import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryBroadcastExec, WindowSortLimitExec}
-import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, SortAggregateExec}
-import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastRangeExchangeExec, ShuffleExchangeExec}
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, DynamicPruningExpression, Expression}
+import org.apache.spark.sql.execution.{FileSourceScanExec, InputAdapter, InSubqueryExec, SparkPlan, SubqueryBroadcastExec, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.exchange.Exchange
+import org.apache.spark.sql.execution.joins.BaseJoinExec
 import org.apache.spark.sql.execution.window.WindowExec
 import org.apache.spark.sql.types.{BooleanType, DateType, NumericType, StringType, StructField}
 
@@ -83,8 +84,8 @@ case class AnalyzeViewCommand(view: TableIdentifier) extends RunnableCommand {
   private def generatePredicate(field: StructField): String = {
     val name = field.name
     field.dataType match {
-      case _: NumericType => s"$name > 1"
-      case _: StringType => s"$name = 'xxx'"
+      case _: NumericType => s"$name != 1"
+      case _: StringType => s"$name != 'xxx'"
       case _: BooleanType => s"$name = boolean('true')"
       case _: DateType => s"$name >= '2000-01-01'"
       case _ => s"$name is not null"
@@ -155,15 +156,31 @@ case class AnalyzeViewCommand(view: TableIdentifier) extends RunnableCommand {
             e
           case e: Expression => e
         }
-      case _: ShuffleExchangeExec | _: BroadcastExchangeExec | _: BroadcastRangeExchangeExec =>
+        totalOperator = totalOperator + 1
+      case _: Exchange =>
         exchangeCount = exchangeCount + 1
-      case _: SortMergeJoinExec | _: ShuffledHashJoinExec | _: BroadcastHashJoinExec =>
+        totalOperator = totalOperator + 1
+      case _: BaseJoinExec =>
         joinCount = joinCount + 1
-      case _: WindowExec | _: WindowSortLimitExec =>
+        totalOperator = totalOperator + 1
+      case _: WindowExec =>
         windowCount = windowCount + 1
-      case _: HashAggregateExec | _: SortAggregateExec =>
+        totalOperator = totalOperator + 1
+      case _: BaseAggregateExec =>
         aggCount = aggCount + 1
-      case _ => totalOperator = totalOperator + 1
+        totalOperator = totalOperator + 1
+      case _: WholeStageCodegenExec | _: InputAdapter =>
+      case aqe: AdaptiveSparkPlanExec =>
+        val subRes = collectPartitionPredicatesAndOperators(aqe)
+        subRes._1.foreach(t => partitionColumnPredicate.put(t._1, t._2))
+        if (subRes._2.length == 5) {
+          totalOperator = totalOperator + subRes._2.head._2 + 1
+          joinCount = joinCount + subRes._2(1)._2
+          exchangeCount = exchangeCount + subRes._2(2)._2
+          aggCount = aggCount + subRes._2(3)._2
+          windowCount = windowCount + subRes._2(4)._2
+        }
+      case _: SparkPlan => totalOperator = totalOperator + 1
     }
     val info = Seq(("Total", totalOperator), ("Join", joinCount), ("Exchange",
       exchangeCount), ("Aggregation", aggCount), ("Window", windowCount))