[KYUUBI #2650] Add FilteredShowColumnsCommand to AuthZ module

### _Why are the changes needed?_

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2650 from zhangrenhua/feature/AddFilteredShowColumnsCommand.

Closes #2650

842a00e [zhangrenhua] Add FilteredShowColumnsCommand to AuthZ module

Authored-by: zhangrenhua <zhangrenhuaman@senses-ai.com>
Signed-off-by: Kent Yao <yao@apache.org>

Author: zhangrenhua

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

@@ -22,7 +22,7 @@ import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.{RunnableCommand, ShowColumnsCommand}
 
 import org.apache.kyuubi.plugin.spark.authz.{ObjectType, OperationType}
 import org.apache.kyuubi.plugin.spark.authz.util.{AuthZUtils, ObjectFilterPlaceHolder, WithInternalChild}
@@ -37,6 +37,8 @@ class RuleReplaceShowObjectCommands extends Rule[LogicalPlan] {
       ObjectFilterPlaceHolder(n)
     case r: RunnableCommand if r.nodeName == "ShowFunctionsCommand" =>
       FilteredShowFunctionsCommand(r)
+    case r: RunnableCommand if r.nodeName == "ShowColumnsCommand" =>
+      FilteredShowColumnsCommand(r)
     case _ => plan
   }
 }
@@ -101,3 +103,24 @@ case class FilteredShowFunctionsCommand(delegated: RunnableCommand)
     result != null && result.getIsAllowed
   }
 }
+
+case class FilteredShowColumnsCommand(delegated: RunnableCommand)
+  extends FilteredShowObjectCommand(delegated) with WithInternalChild {
+
+  override val output: Seq[Attribute] = delegated.output
+
+  override def run(spark: SparkSession): Seq[Row] = {
+    val rows = delegated.run(spark)
+    val table = delegated.asInstanceOf[ShowColumnsCommand].tableName
+    val ugi = AuthZUtils.getAuthzUgi(spark.sparkContext)
+    rows.filter(f =>
+      isAllowed(Row(table.database.orNull, table.table, f.getString(0)), ugi))
+  }
+
+  override protected def isAllowed(r: Row, ugi: UserGroupInformation): Boolean = {
+    val resource = AccessResource(ObjectType.COLUMN, r.getString(0), r.getString(1), r.getString(2))
+    val request = AccessRequest(resource, ugi, OperationType.SHOWCOLUMNS, AccessType.USE)
+    val result = SparkRangerAdminPlugin.isAccessAllowed(request)
+    result != null && result.getIsAllowed
+  }
+}

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

@@ -316,6 +316,26 @@ abstract class RangerSparkExtensionSuite extends KyuubiFunSuite with SparkSessio
       doAs("admin", sql(s"DROP DATABASE IF EXISTS $db3"))
     }
   }
+
+  test("show columns") {
+    val db = "default"
+    val table = "src"
+    val col = "key"
+    val create = s"CREATE TABLE IF NOT EXISTS $db.$table ($col int, value int) USING $format"
+    try {
+      doAs("admin", sql(create))
+
+      doAs("admin", assert(sql(s"SHOW COLUMNS IN $table").count() == 2))
+      doAs("admin", assert(sql(s"SHOW COLUMNS IN $db.$table").count() == 2))
+      doAs("admin", assert(sql(s"SHOW COLUMNS IN $table IN $db").count() == 2))
+
+      doAs("kent", assert(sql(s"SHOW COLUMNS IN $table").count() == 1))
+      doAs("kent", assert(sql(s"SHOW COLUMNS IN $db.$table").count() == 1))
+      doAs("kent", assert(sql(s"SHOW COLUMNS IN $table IN $db").count() == 1))
+    } finally {
+      doAs("admin", sql(s"DROP TABLE IF EXISTS $db.$table"))
+    }
+  }
 }
 
 class InMemoryCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {