[KYUUBI #5793][AUTHZ][BUG] PVM with nested scalar-subquery should not check src table privilege

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

@@ -28,13 +28,27 @@ import org.apache.kyuubi.plugin.spark.authz.rule.permanentview.PermanentViewMark
  * Transforming up [[PermanentViewMarker]]
  */
 class RuleEliminatePermanentViewMarker(sparkSession: SparkSession) extends Rule[LogicalPlan] {
+
+  def eliminatePVM(plan: LogicalPlan): LogicalPlan = {
+    plan.transformUp {
+      case pvm: PermanentViewMarker =>
+        val ret = pvm.child.transformAllExpressions {
+          case s: SubqueryExpression => s.withNewPlan(eliminatePVM(s.plan))
+        }
+        // For each SubqueryExpression's PVM, we should mark as resolved to
+        // avoid check privilege of PVM's internal Subquery.
+        Authorization.markAuthChecked(ret)
+        ret
+    }
+  }
+
   override def apply(plan: LogicalPlan): LogicalPlan = {
     var matched = false
     val eliminatedPVM = plan.transformUp {
       case pvm: PermanentViewMarker =>
         matched = true
         pvm.child.transformAllExpressions {
-          case s: SubqueryExpression => s.withNewPlan(apply(s.plan))
+          case s: SubqueryExpression => s.withNewPlan(eliminatePVM(s.plan))
         }
     }
     if (matched) {

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

@@ -17,6 +17,7 @@
 
 package org.apache.kyuubi.plugin.spark.authz.rule.permanentview
 
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, View}
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -32,15 +33,24 @@ import org.apache.kyuubi.plugin.spark.authz.util.AuthZUtils._
  */
 class RuleApplyPermanentViewMarker extends Rule[LogicalPlan] {
 
+  private def resolveSubqueryExpression(
+      plan: LogicalPlan,
+      catalogTable: CatalogTable): LogicalPlan = {
+    plan.transformAllExpressions {
+      case subquery: SubqueryExpression =>
+        subquery.withNewPlan(plan = PermanentViewMarker(
+          resolveSubqueryExpression(subquery.plan, catalogTable),
+          catalogTable))
+    }
+  }
+
   override def apply(plan: LogicalPlan): LogicalPlan = {
     plan mapChildren {
       case p: PermanentViewMarker => p
       case permanentView: View if hasResolvedPermanentView(permanentView) =>
-        val resolved = permanentView.transformAllExpressions {
-          case subquery: SubqueryExpression =>
-            subquery.withNewPlan(plan = PermanentViewMarker(subquery.plan, permanentView.desc))
-        }
-        PermanentViewMarker(resolved, permanentView.desc)
+        PermanentViewMarker(
+          resolveSubqueryExpression(permanentView, permanentView.desc),
+          permanentView.desc)
       case other => apply(other)
     }
   }

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

@@ -1368,4 +1368,50 @@ class HiveCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
       }
     }
   }
+
+  test("[KYUUBI #5793][BUG] PVM with nested scala-subquery should not src table privilege") {
+    val db1 = defaultDb
+    val table1 = "table1"
+    val table2 = "table2"
+    val table3 = "table3"
+    val view1 = "perm_view"
+    withSingleCallEnabled {
+      withCleanTmpResources(
+        Seq(
+          (s"$db1.$table1", "table"),
+          (s"$db1.$table2", "table"),
+          (s"$db1.$table3", "table"),
+          (s"$db1.$view1", "view"))) {
+        doAs(admin, sql(s"CREATE TABLE IF NOT EXISTS $db1.$table1(id int, scope int)"))
+        doAs(
+          admin,
+          sql(
+            s"""
+               | CREATE TABLE IF NOT EXISTS $db1.$table2(
+               |  id int,
+               |  name string,
+               |  age int,
+               |  scope int)
+               | """.stripMargin))
+        doAs(admin, sql(s"CREATE TABLE IF NOT EXISTS $db1.$table3(id int, scope int)"))
+        doAs(
+          admin,
+          sql(
+            s"""
+               |CREATE VIEW $db1.$view1
+               |AS
+               |SELECT id, name, max(scope) as max_scope, sum(age) sum_age
+               |FROM $db1.$table2
+               |WHERE scope in (
+               |    SELECT max(scope) max_scope
+               |    FROM $db1.$table1
+               |    WHERE id IN (SELECT id FROM $db1.$table3)
+               |)
+               |GROUP BY id, name
+               |""".stripMargin))
+
+        checkAnswer(permViewOnlyUser, s"SELECT * FROM $db1.$view1", Array.empty[Row])
+      }
+    }
+  }
 }