(kyuubi) branch master updated: [KYUUBI #7746][AUTHZ] Report nondeterministic permanent views as nondeterministic to InlineCTE

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git


The following commit(s) were added to refs/heads/master by this push:
     new df5aa6c8b9 [KYUUBI #7746][AUTHZ] Report nondeterministic permanent views as nondeterministic to InlineCTE
df5aa6c8b9 is described below

commit df5aa6c8b966d868f5ad2a0d4f5d148ebddc63ba
Author: Jiwon Park <ji...@apache.org>
AuthorDate: Mon Sep 21 00:01:52 2026 +0800

    [KYUUBI #7746][AUTHZ] Report nondeterministic permanent views as nondeterministic to InlineCTE
    
    ### Why are the changes needed?
    
    Fixes #7746.
    
    `PermanentViewMarker` is a `LeafNode`, so `QueryPlan.deterministic` reports `true` for any
    permanent view, even one over `rand()`, `uuid()` or `reflect()`. `InlineCTE` reads that flag and
    inlines a nondeterministic view CTE that is referenced more than once, so each reference
    recomputes the view and, for expressions that draw a fresh value per evaluation, the references
    disagree. Without authz Spark keeps the CTE out of line and shares one `ReusedExchange`.
    
    This overrides `deterministic` to delegate to the wrapped plan. The same flag feeds
    `SubqueryExpression.deterministic`, so subqueries over such a view are classified the same way
    as without authz.
    
    `override lazy val` compiles against both the `lazy val` in Spark 3.5 and the `def` in 4.0+.
    
    ### How was this patch tested?
    
    Two tests in `HiveCatalogRangerSparkExtensionSuite`:
    
    - a `uuid()` view referenced twice through a CTE keeps a `RepartitionByExpression` in the
      optimized plan and a `ReusedExchange` in the executed plan;
    - a `reflect('java.util.UUID', 'randomUUID')` view referenced twice through a CTE yields the same
      value for both references.
    
    Both fail with `PermanentViewMarker` reverted to master and pass with the change, on Spark 3.5.8,
    4.0.3, 4.1.3 and 4.2.0.
    
    ### Was this patch assisted by generative AI tooling?
    
    Assisted-by: Claude Fable 5.1
    
    Closes #7747 from j1wonpark/KYUUBI-7746.
    
    Closes #7746
    
    3052a8633 [Jiwon Park] [KYUUBI #7746][AUTHZ] Report nondeterministic permanent views as nondeterministic to InlineCTE
    
    Authored-by: Jiwon Park <ji...@apache.org>
    Signed-off-by: Cheng Pan <ch...@apache.org>
---
 .../rule/permanentview/PermanentViewMarker.scala   |  5 +++
 .../authz/ranger/RangerSparkExtensionSuite.scala   | 43 +++++++++++++++++++++-
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/PermanentViewMarker.scala b/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/PermanentViewMarker.scala
index cb233b4651..d3688bf469 100644
--- a/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/PermanentViewMarker.scala
+++ b/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/rule/permanentview/PermanentViewMarker.scala
@@ -37,6 +37,11 @@ case class PermanentViewMarker(child: LogicalPlan, catalogTable: CatalogTable)
 
   override def computeStats(): Statistics = child.stats
 
+  // As a LeafNode the marker has no expressions or children, so QueryPlan.deterministic would
+  // report true even for a view over rand() or uuid(). InlineCTE relies on this flag to keep a
+  // nondeterministic CTE out of line when it is referenced more than once.
+  override lazy val deterministic: Boolean = child.deterministic
+
   override def newInstance(): LogicalPlan = {
     val projectList = child.output.map { case attr =>
       Alias(Cast(attr, attr.dataType), attr.name)(explicitMetadata = Some(attr.metadata))
diff --git a/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala b/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala
index 8805f0cd52..4c04cd6843 100644
--- a/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala
+++ b/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala
@@ -28,7 +28,7 @@ import org.apache.spark.SparkConf
 import org.apache.spark.sql.{Row, SparkSessionExtensions}
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
-import org.apache.spark.sql.catalyst.plans.logical.Statistics
+import org.apache.spark.sql.catalyst.plans.logical.{RepartitionByExpression, Statistics}
 import org.apache.spark.sql.execution.columnar.InMemoryRelation
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 
@@ -741,6 +741,47 @@ class HiveCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
     }
   }
 
+  test("a nondeterministic view referenced twice through a CTE is computed once") {
+    val view = s"$defaultDb.pvm_nondeterministic_view"
+    withCleanTmpResources(Seq((view, "view"))) {
+      doAs(admin) {
+        sql(s"CREATE VIEW $view AS SELECT id, uuid() AS r FROM range(20)")
+        val query = sql(s"WITH c AS (SELECT * FROM $view) " +
+          "SELECT a.id, a.r AS l, b.r AS r FROM c a JOIN c b ON a.id = b.id")
+        // Spark keeps a nondeterministic CTE out of line when it is referenced more than once:
+        // each reference is planned under a rebalance shuffle and the physical plan reuses one
+        // Exchange. Inlining it would evaluate uuid() separately for each reference.
+        assert(
+          query.queryExecution.optimizedPlan.exists(_.isInstanceOf[RepartitionByExpression]),
+          "Expected the nondeterministic CTE to stay out of line")
+        val rows = query.collect()
+        assert(rows.length == 20)
+        assert(rows.forall(row => row.getString(1) == row.getString(2)))
+        assert(
+          query.queryExecution.executedPlan.toString.contains("ReusedExchange"),
+          "Expected both CTE references to share one Exchange")
+      }
+    }
+  }
+
+  test("a nondeterministic view referenced twice through a CTE yields the same values") {
+    val view = s"$defaultDb.pvm_reflect_nondeterministic_view"
+    withCleanTmpResources(Seq((view, "view"))) {
+      doAs(admin) {
+        // uuid() and rand() are seeded at analysis time, so inlined copies of the view happen to
+        // agree. reflect() draws a fresh value on every evaluation, so they do not.
+        sql(s"CREATE VIEW $view AS " +
+          "SELECT id, reflect('java.util.UUID', 'randomUUID') AS r FROM range(20)")
+        val rows = sql(s"WITH c AS (SELECT * FROM $view) " +
+          "SELECT a.id, a.r AS l, b.r AS r FROM c a JOIN c b ON a.id = b.id").collect()
+        assert(rows.length == 20)
+        assert(
+          rows.forall(row => row.getString(1) == row.getString(2)),
+          "Expected both CTE references to see the same evaluation of the view")
+      }
+    }
+  }
+
   test("[KYUUBI #3608] Support {OWNER} variable for queries") {
     val db = defaultDb
     val table = "owner_variable"