[KYUUBI #5780][AUTHZ] Treating PermanentViewMarker as LeafNode make c…

…ode simple and got correct privilege object

# 🔍 Description
## Issue References 🔗

This pull request fixes #5780

## Describe Your Solution 🔧

Currently, we convert persist view to PermanentViewMaker,  but after optimizer, it changed its child, making it hard to do column prune and get the right column privilege object of persist view.

In this pr, we change PVM as a LeafNode, then we can directly treat it as a `HiveRelation` since we won't change its internal plan to make code simpler.
But we need to re-optimize the child plan after do privilege check.

## Types of changes 🔖

- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

#### Behavior Without This Pull Request ⚰️
For sql such as  below table and view
```
CREATE TABLE IF NOT EXISTS $db1.$table1(id int, scope int);

CREATE TABLE IF NOT EXISTS $db1.$table2(
id int,
name string,
age int,
scope int)
.stripMargin);

CREATE VIEW $db1.$view1
AS
WITH temp AS (
    SELECT max(scope) max_scope
    FROM $db1.$table1)
SELECT id, name, max(scope) as max_scope, sum(age) sum_age
FROM $db1.$table2
WHERE scope in (SELECT max_scope FROM temp)
GROUP BY id, name
```
When we execute  query on `$db1.$view1`
```
SELECT id as new_id, name, max_scope FROM $db1.$view1
```

It will first execute the subquery in the query,  then got a un-correct column privilege

#### Behavior With This Pull Request 🎉
After this change, since PVM is a LeafNode, we won't execute the subquery under PVM, and we directly got the correct column privilege.

#### Related Unit Tests
Existed UT

---

# Checklists
## 📝 Author Self Checklist

- [x] My code follows the [style guidelines](https://kyuubi.readthedocs.io/en/master/contributing/code/style.html) of this project
- [x] I have performed a self-review
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes
- [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

## 📝 Committer Pre-Merge Checklist

- [x] Pull request title is okay.
- [x] No license issues.
- [x] Milestone correctly set?
- [x] Test coverage is ok
- [x] Assignees are selected.
- [x] Minimum number of approvals
- [x] No changes are requested

**Be nice. Be informative.**

Closes #5781 from AngersZhuuuu/KYUUBI-5780.

Closes #5780

3c18bb7 [Angerszhuuuu] Merge branch 'master' into KYUUBI-5780
64f7947 [Angerszhuuuu] update
b4f6fc0 [Angerszhuuuu] Merge branch 'master' into KYUUBI-5780
fbc989a [Angerszhuuuu] Update Authorization.scala
2113cf5 [Angerszhuuuu] Update RuleApplyPermanentViewMarker.scala
5dbe232 [Angerszhuuuu] Update WithInternalChildren.scala
04a40c3 [Angerszhuuuu] update
57bf5ba [Angerszhuuuu] update
738d506 [Angerszhuuuu] Update RuleApplyPermanentViewMarker.scala
bade427 [Angerszhuuuu] [KYUUBI #5780][AUTHZ] Kyuubi tread PVM ass LeafNode to make logic more simple

Authored-by: Angerszhuuuu <[email protected]>
Signed-off-by: Kent Yao <[email protected]>

extensions/spark/kyuubi-spark-authz/src/main/resources/database_command_spec.json

@@ -215,4 +215,4 @@
   } ],
   "opType" : "SWITCHDATABASE",
   "uriDescs" : [ ]
-} ]
+} ]

extensions/spark/kyuubi-spark-authz/src/main/resources/function_command_spec.json

@@ -111,4 +111,4 @@
     "comment" : ""
   } ],
   "opType" : "RELOADFUNCTION"
-} ]
+} ]

extensions/spark/kyuubi-spark-authz/src/main/resources/scan_command_spec.json

@@ -111,4 +111,4 @@
     "comment" : ""
   } ],
   "uriDescs" : [ ]
-} ]
+} ]

extensions/spark/kyuubi-spark-authz/src/main/resources/table_command_spec.json

@@ -2528,4 +2528,4 @@
     "isInput" : false,
     "comment" : "Delta"
   } ]
-} ]
+} ]

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/PrivilegesBuilder.scala

@@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory
 import org.apache.kyuubi.plugin.spark.authz.OperationType.OperationType
 import org.apache.kyuubi.plugin.spark.authz.PrivilegeObjectActionType._
 import org.apache.kyuubi.plugin.spark.authz.rule.Authorization._
-import org.apache.kyuubi.plugin.spark.authz.rule.permanentview.PermanentViewMarker
 import org.apache.kyuubi.plugin.spark.authz.rule.rowfilter._
 import org.apache.kyuubi.plugin.spark.authz.serde._
 import org.apache.kyuubi.plugin.spark.authz.util.AuthZUtils._
@@ -68,13 +67,7 @@ object PrivilegesBuilder {
 
     def mergeProjection(table: Table, plan: LogicalPlan): Unit = {
       if (projectionList.isEmpty) {
-        plan match {
-          case pvm: PermanentViewMarker
-              if pvm.isSubqueryExpressionPlaceHolder || pvm.output.isEmpty =>
-            privilegeObjects += PrivilegeObject(table, pvm.outputColNames)
-          case _ =>
-            privilegeObjects += PrivilegeObject(table, plan.output.map(_.name))
-        }
+        privilegeObjects += PrivilegeObject(table, plan.output.map(_.name))
       } else {
         val cols = (projectionList ++ conditionList).flatMap(collectLeaves)
           .filter(plan.outputSet.contains).map(_.name).distinct

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtension.scala

@@ -53,7 +53,7 @@ class RangerSparkExtension extends (SparkSessionExtensions => Unit) {
     v1.injectResolutionRule(RuleApplyDataMaskingStage1)
     v1.injectOptimizerRule(_ => new RuleEliminateMarker())
     v1.injectOptimizerRule(new RuleAuthorization(_))
-    v1.injectOptimizerRule(_ => new RuleEliminatePermanentViewMarker())
+    v1.injectOptimizerRule(new RuleEliminatePermanentViewMarker(_))
     v1.injectOptimizerRule(_ => new RuleEliminateTypeOf())
     v1.injectPlannerStrategy(new FilterDataSourceV2Strategy(_))
   }

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/Authorization.scala

@@ -18,13 +18,12 @@
 package org.apache.kyuubi.plugin.spark.authz.rule
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Subquery}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, View}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.execution.SQLExecution.EXECUTION_ID_KEY
 
 import org.apache.kyuubi.plugin.spark.authz.rule.Authorization._
-import org.apache.kyuubi.plugin.spark.authz.rule.permanentview.PermanentViewMarker
 import org.apache.kyuubi.plugin.spark.authz.util.ReservedKeys._
 
 abstract class Authorization(spark: SparkSession) extends Rule[LogicalPlan] {
@@ -51,21 +50,18 @@ object Authorization {
     }
   }
 
-  protected def markAuthChecked(plan: LogicalPlan): LogicalPlan = {
+  def markAuthChecked(plan: LogicalPlan): LogicalPlan = {
     plan.setTagValue(KYUUBI_AUTHZ_TAG, ())
     plan transformDown {
-      case pvm: PermanentViewMarker =>
-        markAllNodesAuthChecked(pvm)
-      case subquery: Subquery =>
-        markAllNodesAuthChecked(subquery)
+      // TODO: Add this line Support for spark3.1, we can remove this
+      //       after spark 3.2 since https://issues.apache.org/jira/browse/SPARK-34269
+      case view: View =>
+        markAllNodesAuthChecked(view.child)
     }
   }
 
   protected def isAuthChecked(plan: LogicalPlan): Boolean = {
-    plan match {
-      case subquery: Subquery => isAuthChecked(subquery.child)
-      case p => p.getTagValue(KYUUBI_AUTHZ_TAG).nonEmpty
-    }
+    plan.getTagValue(KYUUBI_AUTHZ_TAG).nonEmpty
   }
 
   def setExplainCommandExecutionId(sparkSession: SparkSession): Unit = {

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/RuleEliminatePermanentViewMarker.scala

@@ -17,6 +17,7 @@
 
 package org.apache.kyuubi.plugin.spark.authz.rule
 
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -26,12 +27,21 @@ import org.apache.kyuubi.plugin.spark.authz.rule.permanentview.PermanentViewMark
 /**
  * Transforming up [[PermanentViewMarker]]
  */
-class RuleEliminatePermanentViewMarker extends Rule[LogicalPlan] {
+class RuleEliminatePermanentViewMarker(sparkSession: SparkSession) extends Rule[LogicalPlan] {
   override def apply(plan: LogicalPlan): LogicalPlan = {
-    plan.transformUp {
-      case pvm: PermanentViewMarker => pvm.child.transformAllExpressions {
+    var matched = false
+    val eliminatedPVM = plan.transformUp {
+      case pvm: PermanentViewMarker =>
+        matched = true
+        pvm.child.transformAllExpressions {
           case s: SubqueryExpression => s.withNewPlan(apply(s.plan))
         }
     }
+    if (matched) {
+      Authorization.markAuthChecked(eliminatedPVM)
+      sparkSession.sessionState.optimizer.execute(eliminatedPVM)
+    } else {
+      eliminatedPVM
+    }
   }
 }

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/PermanentViewMarker.scala

@@ -19,20 +19,9 @@ package org.apache.kyuubi.plugin.spark.authz.rule.permanentview
 
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
 
-import org.apache.kyuubi.plugin.spark.authz.util.WithInternalChild
-
-case class PermanentViewMarker(
-    child: LogicalPlan,
-    catalogTable: CatalogTable,
-    outputColNames: Seq[String],
-    isSubqueryExpressionPlaceHolder: Boolean = false) extends UnaryNode
-  with WithInternalChild {
+case class PermanentViewMarker(child: LogicalPlan, catalogTable: CatalogTable) extends LeafNode {
 
   override def output: Seq[Attribute] = child.output
-
-  override def withNewChildInternal(newChild: LogicalPlan): LogicalPlan =
-    copy(child = newChild)
-
 }

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/RuleApplyPermanentViewMarker.scala

@@ -38,14 +38,9 @@ class RuleApplyPermanentViewMarker extends Rule[LogicalPlan] {
       case permanentView: View if hasResolvedPermanentView(permanentView) =>
         val resolved = permanentView.transformAllExpressions {
           case subquery: SubqueryExpression =>
-            subquery.withNewPlan(plan =
-              PermanentViewMarker(
-                subquery.plan,
-                permanentView.desc,
-                permanentView.output.map(_.name),
-                true))
+            subquery.withNewPlan(plan = PermanentViewMarker(subquery.plan, permanentView.desc))
         }
-        PermanentViewMarker(resolved, resolved.desc, resolved.output.map(_.name))
+        PermanentViewMarker(resolved, permanentView.desc)
       case other => apply(other)
     }
   }

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala

@@ -889,7 +889,7 @@ class HiveCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
             sql(s"SELECT id as new_id, name, max_scope FROM $db1.$view1".stripMargin).show()))
         assert(e2.getMessage.contains(
           s"does not have [select] privilege on " +
-            s"[$db1/$view1/id,$db1/$view1/name,$db1/$view1/max_scope,$db1/$view1/sum_age]"))
+            s"[$db1/$view1/id,$db1/$view1/name,$db1/$view1/max_scope]"))
       }
     }
   }