Skip to content

[KYUUBI #4617] [AUTHZ] Collect results for filtered show objects ahead to prevent holding unserializable spark plan #4634

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[KYUUBI #4617] [AUTHZ] Collect results for filtered show objects ahead to prevent holding unserializable spark plan #4634

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
65ce03a
fix 4617
bowenliang123 Mar 30, 2023
fe00ef5
rename results to result
bowenliang123 Mar 31, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...c/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/FilterDataSourceV2Strategy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,13 @@ import org.apache.kyuubi.plugin.spark.authz.util.ObjectFilterPlaceHolder
class FilterDataSourceV2Strategy(spark: SparkSession) extends Strategy {
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case ObjectFilterPlaceHolder(child) if child.nodeName == "ShowNamespaces" =>
spark.sessionState.planner.plan(child).map(FilteredShowNamespaceExec).toSeq
spark.sessionState.planner.plan(child)
.map(FilteredShowNamespaceExec(_, spark.sparkContext)).toSeq

case ObjectFilterPlaceHolder(child) if child.nodeName == "ShowTables" =>
spark.sessionState.planner.plan(child).map(FilteredShowTablesExec).toSeq
spark.sessionState.planner.plan(child)
.map(FilteredShowTablesExec(_, spark.sparkContext)).toSeq

case _ => Nil
}
}
36 changes: 25 additions & 11 deletions .../src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/FilteredShowObjectsExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kyuubi.plugin.spark.authz.ranger

import org.apache.hadoop.security.UserGroupInformation
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Attribute
Expand All @@ -26,24 +27,29 @@ import org.apache.kyuubi.plugin.spark.authz.{ObjectType, OperationType}
import org.apache.kyuubi.plugin.spark.authz.util.AuthZUtils

trait FilteredShowObjectsExec extends LeafExecNode {
def delegated: SparkPlan
def result: Array[InternalRow]

final override def output: Seq[Attribute] = delegated.output

final private lazy val result = {
delegated.executeCollect().filter(isAllowed(_, AuthZUtils.getAuthzUgi(sparkContext)))
}
override def output: Seq[Attribute]

final override def doExecute(): RDD[InternalRow] = {
sparkContext.parallelize(result, 1)
}
}

protected def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean
trait FilteredShowObjectsCheck {
def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean
}

case class FilteredShowNamespaceExec(delegated: SparkPlan) extends FilteredShowObjectsExec {
case class FilteredShowNamespaceExec(result: Array[InternalRow], output: Seq[Attribute])
extends FilteredShowObjectsExec {}
object FilteredShowNamespaceExec extends FilteredShowObjectsCheck {
def apply(delegated: SparkPlan, sc: SparkContext): FilteredShowNamespaceExec = {
val result = delegated.executeCollect()
.filter(isAllowed(_, AuthZUtils.getAuthzUgi(sc)))
new FilteredShowNamespaceExec(result, delegated.output)
}

override protected def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean = {
override def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean = {
val database = r.getString(0)
val resource = AccessResource(ObjectType.DATABASE, database, null, null)
val request = AccessRequest(resource, ugi, OperationType.SHOWDATABASES, AccessType.USE)
Expand All @@ -52,8 +58,16 @@ case class FilteredShowNamespaceExec(delegated: SparkPlan) extends FilteredShowO
}
}

case class FilteredShowTablesExec(delegated: SparkPlan) extends FilteredShowObjectsExec {
override protected def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean = {
case class FilteredShowTablesExec(result: Array[InternalRow], output: Seq[Attribute])
extends FilteredShowObjectsExec {}
object FilteredShowTablesExec extends FilteredShowObjectsCheck {
def apply(delegated: SparkPlan, sc: SparkContext): FilteredShowNamespaceExec = {
val result = delegated.executeCollect()
.filter(isAllowed(_, AuthZUtils.getAuthzUgi(sc)))
new FilteredShowNamespaceExec(result, delegated.output)
}

override def isAllowed(r: InternalRow, ugi: UserGroupInformation): Boolean = {
val database = r.getString(0)
val table = r.getString(1)
val isTemp = r.getBoolean(2)
Expand Down
2 changes: 2 additions & 0 deletions ...rc/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,7 @@ abstract class RangerSparkExtensionSuite extends AnyFunSuite
doAs("admin", assert(sql(s"show tables from $db").collect().length === 2))
doAs("bob", assert(sql(s"show tables from $db").collect().length === 0))
doAs("i_am_invisible", assert(sql(s"show tables from $db").collect().length === 0))
doAs("i_am_invisible", assert(sql(s"show tables from $db").limit(1).isEmpty))
}
}

Expand All @@ -247,6 +248,7 @@ abstract class RangerSparkExtensionSuite extends AnyFunSuite

doAs("bob", assert(sql(s"SHOW DATABASES").collect().length == 1))
doAs("bob", assert(sql(s"SHOW DATABASES").collectAsList().get(0).getString(0) == "default"))
doAs("i_am_invisible", assert(sql(s"SHOW DATABASES").limit(1).isEmpty))
}
}

Expand Down