[KYUUBI #7198] Remove support for Spark 3.2

### Why are the changes needed?

Discussion in mailing list: https://lists.apache.org/thread/qrbknzjj3jcwjc9nd64qmtohrwfk1kjp

Spark 3.2 is EOL on 2023/04, Kyuubi deprecated support for Spark 3.2 in 1.10 #6545

This PR proposes to fully drop support for Spark 3.2.

### How was this patch tested?

Pass GHA.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #7198 from pan3793/remove-spark-3.2.

Closes #7198

7d7e21c [Cheng Pan] address comments
219fb79 [Cheng Pan] fix
b4c8ae7 [Cheng Pan] fix
6ffd9c2 [Cheng Pan] address comments
1d0dc2d [Cheng Pan] Update extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParserHelperSuite.scala
343b658 [Cheng Pan] Remove support of Spark 3.2

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

Author: pan3793

.github/workflows/master.yml

@@ -60,11 +60,6 @@ jobs:
             spark-archive: '-Pscala-2.13'
             exclude-tags: ''
             comment: 'normal'
-          - java: 8
-            spark: '3.5'
-            spark-archive: '-Dspark.archive.mirror=https://archive.apache.org/dist/spark/spark-3.2.4 -Dspark.archive.name=spark-3.2.4-bin-hadoop3.2.tgz -Pzookeeper-3.6'
-            exclude-tags: '-Dmaven.plugin.scalatest.exclude.tags=org.scalatest.tags.Slow,org.apache.kyuubi.tags.DeltaTest,org.apache.kyuubi.tags.IcebergTest,org.apache.kyuubi.tags.PaimonTest,org.apache.kyuubi.tags.HudiTest,org.apache.kyuubi.tags.SparkLocalClusterTest'
-            comment: 'verify-on-spark-3.2-binary'
           - java: 8
             spark: '3.5'
             spark-archive: '-Dspark.archive.mirror=https://archive.apache.org/dist/spark/spark-3.3.3 -Dspark.archive.name=spark-3.3.3-bin-hadoop3.tgz -Pzookeeper-3.6'

docs/deployment/migration-guide.md

@@ -19,6 +19,7 @@
 
 ## Upgrading from Kyuubi 1.10 to 1.11
 
+* Since Kyuubi 1.11, the support of Spark engine for Spark 3.2 is removed.
 * Since Kyuubi 1.11, the support of Flink engine for Flink 1.17 and 1.18 are deprecated, and will be removed in the future.
 * Since Kyuubi 1.11, the configuration `spark.sql.watchdog.forcedMaxOutputRows` provided by Kyuubi Spark extension is removed, consider using `kyuubi.operation.result.max.rows` instead. Note, the latter works without requirement of installing Kyuubi Spark extension.
 * Since Kyuubi 1.11, if the engine is running in cluster mode, Kyuubi will respect the `kyuubi.session.engine.startup.waitCompletion` config to determine whether to wait for the engine completion or not. If the engine is running in client mode, Kyuubi will always wait for the engine completion. And for Spark engine, Kyuubi will append the `spark.yarn.submit.waitAppCompletion` and `spark.kubernetes.submission.waitAppCompletion` configs to the engine conf based on the value of `kyuubi.session.engine.startup.waitCompletion`.

docs/quick_start/quick_start.rst

@@ -43,7 +43,7 @@ pre-installed and the ``JAVA_HOME`` is correctly set to each component.
   **Kyuubi**       Gateway      \ |release| \        - Kyuubi Server
                    Engine lib                        - Kyuubi Engine
                    Beeline                           - Kyuubi Beeline
-  **Spark**        Engine       3.2 to 3.5, 4.0      A Spark distribution
+  **Spark**        Engine       3.3 to 3.5, 4.0      A Spark distribution
   **Flink**        Engine       1.17 to 1.20         A Flink distribution
   **Trino**        Engine       N/A                  A Trino cluster allows to access via trino-client v411
   **Doris**        Engine       N/A                  A Doris cluster

extensions/spark/kyuubi-spark-authz/README.md

@@ -37,8 +37,8 @@ build/mvn clean package -DskipTests -pl :kyuubi-spark-authz_2.12 -am -Dspark.ver
 - [x] 3.5.x (default)
 - [x] 3.4.x
 - [x] 3.3.x
-- [x] 3.2.x
-- [x] 3.1.x
+- [ ] 3.2.x
+- [ ] 3.1.x
 - [ ] 3.0.x
 - [ ] 2.4.x and earlier
 

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

@@ -840,17 +840,6 @@
     "isInput" : true,
     "comment" : ""
   } ]
-}, {
-  "classname" : "org.apache.spark.sql.execution.command.AddFileCommand",
-  "tableDescs" : [ ],
-  "opType" : "ADD",
-  "queryDescs" : [ ],
-  "uriDescs" : [ {
-    "fieldName" : "path",
-    "fieldExtractor" : "StringURIExtractor",
-    "isInput" : true,
-    "comment" : ""
-  } ]
 }, {
   "classname" : "org.apache.spark.sql.execution.command.AddFilesCommand",
   "tableDescs" : [ ],
@@ -969,22 +958,6 @@
   "opType" : "ALTERTABLE_DROPPARTS",
   "queryDescs" : [ ],
   "uriDescs" : [ ]
-}, {
-  "classname" : "org.apache.spark.sql.execution.command.AlterTableRecoverPartitionsCommand",
-  "tableDescs" : [ {
-    "fieldName" : "tableName",
-    "fieldExtractor" : "TableIdentifierTableExtractor",
-    "columnDesc" : null,
-    "actionTypeDesc" : null,
-    "tableTypeDesc" : null,
-    "catalogDesc" : null,
-    "isInput" : false,
-    "setCurrentDatabaseIfMissing" : false,
-    "comment" : ""
-  } ],
-  "opType" : "MSCK",
-  "queryDescs" : [ ],
-  "uriDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.execution.command.AlterTableRenameCommand",
   "tableDescs" : [ {

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

@@ -224,24 +224,7 @@ object PrivilegesBuilder {
           }
         }
         spec.queries(plan).foreach { p =>
-          if (p.resolved) {
-            buildQuery(Project(p.output, p), inputObjs, spark = spark)
-          } else {
-            try {
-              // For spark 3.1, Some command such as CreateTableASSelect, its query was unresolved,
-              // Before this pr, we just ignore it, now we support this.
-              val analyzed = spark.sessionState.analyzer.execute(p)
-              buildQuery(Project(analyzed.output, analyzed), inputObjs, spark = spark)
-            } catch {
-              case e: Exception =>
-                LOG.debug(
-                  s"""
-                     |Failed to analyze unresolved
-                     |$p
-                     |due to ${e.getMessage}""".stripMargin,
-                  e)
-            }
-          }
+          buildQuery(Project(p.output, p), inputObjs, spark = spark)
         }
         spec.operationType
 

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

@@ -18,7 +18,7 @@
 package org.apache.kyuubi.plugin.spark.authz.rule
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, View}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 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
@@ -52,12 +52,7 @@ object Authorization {
 
   def markAuthChecked(plan: LogicalPlan): LogicalPlan = {
     plan.setTagValue(KYUUBI_AUTHZ_TAG, ())
-    plan transformDown {
-      // 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)
-    }
+    plan
   }
 
   protected def isAuthChecked(plan: LogicalPlan): Boolean = {

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

@@ -17,18 +17,11 @@
 package org.apache.kyuubi.plugin.spark.authz.rule.rowfilter
 
 import org.apache.spark.sql.{SparkSession, Strategy}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.SparkPlan
 
 case class FilterDataSourceV2Strategy(spark: SparkSession) extends Strategy {
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
-    // For Spark 3.1 and below, `ColumnPruning` rule will set `ObjectFilterPlaceHolder#child` to
-    // `Project`
-    case ObjectFilterPlaceHolder(Project(_, child)) if child.nodeName == "ShowNamespaces" =>
-      spark.sessionState.planner.plan(child)
-        .map(FilteredShowNamespaceExec(_, spark.sparkContext)).toSeq
-
-    // For Spark 3.2 and above
     case ObjectFilterPlaceHolder(child) if child.nodeName == "ShowNamespaces" =>
       spark.sessionState.planner.plan(child)
         .map(FilteredShowNamespaceExec(_, spark.sparkContext)).toSeq

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

@@ -84,8 +84,6 @@ private[authz] object AuthZUtils {
   }
 
   lazy val SPARK_RUNTIME_VERSION: SemanticVersion = SemanticVersion(SPARK_VERSION)
-  lazy val isSparkV32OrGreater: Boolean = SPARK_RUNTIME_VERSION >= "3.2"
-  lazy val isSparkV33OrGreater: Boolean = SPARK_RUNTIME_VERSION >= "3.3"
   lazy val isSparkV34OrGreater: Boolean = SPARK_RUNTIME_VERSION >= "3.4"
   lazy val isSparkV35OrGreater: Boolean = SPARK_RUNTIME_VERSION >= "3.5"
   lazy val isSparkV40OrGreater: Boolean = SPARK_RUNTIME_VERSION >= "4.0"

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

@@ -244,41 +244,6 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
     assert(accessType === AccessType.ALTER)
   }
 
-  test("AlterTableRecoverPartitionsCommand") {
-    // AlterTableRecoverPartitionsCommand exists in the version below 3.2
-    assume(!isSparkV32OrGreater)
-    val tableName = reusedDb + "." + "TableToMsck"
-    withTable(tableName) { _ =>
-      sql(
-        s"""
-           |CREATE TABLE $tableName
-           |(key int, value string, pid string)
-           |USING parquet
-           |PARTITIONED BY (pid)""".stripMargin)
-      val sqlStr =
-        s"""
-           |MSCK REPAIR TABLE $tableName
-           |""".stripMargin
-      val plan = sql(sqlStr).queryExecution.analyzed
-      val (inputs, outputs, operationType) = PrivilegesBuilder.build(plan, spark)
-      assert(operationType === MSCK)
-      assert(inputs.isEmpty)
-
-      assert(outputs.size === 1)
-      outputs.foreach { po =>
-        assert(po.actionType === PrivilegeObjectActionType.OTHER)
-        assert(po.privilegeObjectType === PrivilegeObjectType.TABLE_OR_VIEW)
-        assert(po.catalog.isEmpty)
-        assertEqualsIgnoreCase(reusedDb)(po.dbname)
-        assertEqualsIgnoreCase(tableName.split("\\.").last)(po.objectName)
-        assert(po.columns.isEmpty)
-        checkTableOwner(po)
-        val accessType = ranger.AccessType(po, operationType, isInput = false)
-        assert(accessType === AccessType.ALTER)
-      }
-    }
-  }
-
   // ALTER TABLE default.StudentInfo PARTITION (age='10') RENAME TO PARTITION (age='15');
   test("AlterTableRenamePartitionCommand") {
     sql(s"ALTER TABLE $reusedPartTable ADD IF NOT EXISTS PARTITION (pid=1)")
@@ -367,11 +332,8 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
     assert(po0.privilegeObjectType === PrivilegeObjectType.TABLE_OR_VIEW)
     assertEqualsIgnoreCase(reusedDb)(po0.dbname)
     assertEqualsIgnoreCase(reusedPartTableShort)(po0.objectName)
-    if (isSparkV32OrGreater) {
-      // Query in AlterViewAsCommand can not be resolved before SPARK-34698
-      assert(po0.columns === Seq("key", "pid", "value"))
-      checkTableOwner(po0)
-    }
+    assert(po0.columns === Seq("key", "pid", "value"))
+    checkTableOwner(po0)
     val accessType0 = ranger.AccessType(po0, operationType, isInput = true)
     assert(accessType0 === AccessType.SELECT)
 
@@ -482,7 +444,6 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
   }
 
   test("AnalyzeTablesCommand") {
-    assume(isSparkV32OrGreater)
     val plan = sql(s"ANALYZE TABLES IN $reusedDb COMPUTE STATISTICS")
       .queryExecution.analyzed
     val (in, out, operationType) = PrivilegesBuilder.build(plan, spark)
@@ -626,7 +587,7 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
     assert(po.actionType === PrivilegeObjectActionType.OTHER)
     assert(po.privilegeObjectType === PrivilegeObjectType.FUNCTION)
     assert(po.catalog.isEmpty)
-    val db = if (isSparkV33OrGreater) defaultDb else null
+    val db = defaultDb
     assertEqualsIgnoreCase(db)(po.dbname)
     assertEqualsIgnoreCase("CreateFunctionCommand")(po.objectName)
     assert(po.columns.isEmpty)
@@ -658,7 +619,7 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
     assert(po.actionType === PrivilegeObjectActionType.OTHER)
     assert(po.privilegeObjectType === PrivilegeObjectType.FUNCTION)
     assert(po.catalog.isEmpty)
-    val db = if (isSparkV33OrGreater) defaultDb else null
+    val db = defaultDb
     assertEqualsIgnoreCase(db)(po.dbname)
     assertEqualsIgnoreCase("DropFunctionCommand")(po.objectName)
     assert(po.columns.isEmpty)
@@ -678,7 +639,7 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
     assert(po.actionType === PrivilegeObjectActionType.OTHER)
     assert(po.privilegeObjectType === PrivilegeObjectType.FUNCTION)
     assert(po.catalog.isEmpty)
-    val db = if (isSparkV33OrGreater) defaultDb else null
+    val db = defaultDb
     assertEqualsIgnoreCase(db)(po.dbname)
     assertEqualsIgnoreCase("RefreshFunctionCommand")(po.objectName)
     assert(po.columns.isEmpty)
@@ -927,8 +888,6 @@ abstract class PrivilegesBuilderSuite extends AnyFunSuite
   }
 
   test("RepairTableCommand") {
-    // only spark 3.2 or greater has RepairTableCommand
-    assume(isSparkV32OrGreater)
     val tableName = reusedDb + "." + "TableToRepair"
     withTable(tableName) { _ =>
       sql(