[GLUTEN-11550][UT] Add GlutenTestSetWithSystemPropertyTrait, enable 6 suites, remove 7 TestHiveSingleton suites, fix JobTagging flaky test

gluten-ut/common/src/test/scala/org/apache/spark/sql/GlutenTestSetWithSystemPropertyTrait.scala

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql
+
+/**
+ * A test trait for suites whose parent creates per-test SparkContext/SparkSession (via
+ * LocalSparkContext, LocalSparkSession, or direct `new SparkContext`).
+ *
+ * Injects GlutenPlugin config via system properties so per-test sessions inherit it. SparkConf
+ * reads system properties prefixed with "spark." as defaults during construction.
+ *
+ * Do NOT use GlutenTestsTrait or GlutenSQLTestsTrait for these suites -- they create a shared
+ * SparkSession in beforeAll() that conflicts with per-test SparkContext creation.
+ */
+trait GlutenTestSetWithSystemPropertyTrait extends GlutenTestsCommonTrait {
+
+  override def beforeAll(): Unit = {
+    System.setProperty("spark.plugins", "org.apache.gluten.GlutenPlugin")
+    System.setProperty("spark.memory.offHeap.enabled", "true")
+    System.setProperty("spark.memory.offHeap.size", "1024MB")
+    System.setProperty(
+      "spark.shuffle.manager",
+      "org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+    super.beforeAll()
+  }
+
+  override def afterAll(): Unit = {
+    super.afterAll()
+    System.clearProperty("spark.plugins")
+    System.clearProperty("spark.memory.offHeap.enabled")
+    System.clearProperty("spark.memory.offHeap.size")
+    System.clearProperty("spark.shuffle.manager")
+  }
+}

gluten-ut/excluded-spark-uts.md

@@ -3,4 +3,5 @@
 
 | Suites                                                                                                                                                                               | Versions | Reason to exclude                                                                                                                                                                                                                                                                                                                                                                                |
 |--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| org.apache.spark.sql.SingleLevelAggregateHashMapSuite<br/>org.apache.spark.sql.TwoLevelAggregateHashMapSuite<br/>org.apache.spark.sql.TwoLevelAggregateHashMapWithVectorizedMapSuite | 4.0, 4.1 | This UTs is similar to `DataFrameAggregateSuite`.<br/>The only difference being that it contains variation for Spark codegen enabled and disabled and enabling single level or two level aggregate hash-maps which is also specific to Spark Aggregate implementation.<br/>We already run `GlutenDataFrameAggregateSuite` for `DataFrameAggregateSuite` so these additional suites doesn't need to run |
+| org.apache.spark.sql.SingleLevelAggregateHashMapSuite<br/>org.apache.spark.sql.TwoLevelAggregateHashMapSuite<br/>org.apache.spark.sql.TwoLevelAggregateHashMapWithVectorizedMapSuite | 4.0, 4.1 | This UTs is similar to `DataFrameAggregateSuite`.<br/>The only difference being that it contains variation for Spark codegen enabled and disabled and enabling single level or two level aggregate hash-maps which is also specific to Spark Aggregate implementation.<br/>We already run `GlutenDataFrameAggregateSuite` for `DataFrameAggregateSuite` so these additional suites doesn't need to run |
+| org.apache.spark.sql.sources.GlutenBucketedReadWithHiveSupportSuite<br/>org.apache.spark.sql.sources.GlutenBucketedWriteWithHiveSupportSuite<br/>org.apache.spark.sql.sources.GlutenDisableUnnecessaryBucketedScanWithHiveSupportSuite<br/>org.apache.spark.sql.sources.GlutenCommitFailureTestRelationSuite<br/>org.apache.spark.sql.sources.GlutenJsonHadoopFsRelationSuite<br/>org.apache.spark.sql.sources.GlutenParquetHadoopFsRelationSuite<br/>org.apache.spark.sql.sources.GlutenSimpleTextHadoopFsRelationSuite | 4.0, 4.1 | These suites extend `TestHiveSingleton` which provides a global HiveSession singleton. The `TestHiveSingleton.spark` (protected val) conflicts with Gluten test traits' `spark` (implicit protected def), making it impossible to load GlutenPlugin through any Gluten test trait. The non-Hive variants of these suites (e.g. `GlutenBucketedReadWithoutHiveSupportSuite`) are already enabled and cover the same functionality. |

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -680,7 +680,7 @@ class VeloxTestSettings extends BackendTestSettings {
   // TODO: 4.x enableSuite[GlutenDataSourceScanExecRedactionSuite]  // 2 failures
   // TODO: 4.x enableSuite[GlutenDataSourceV2ScanExecRedactionSuite]  // 2 failures
   enableSuite[GlutenExecuteImmediateEndToEndSuite]
-  // TODO: 4.x enableSuite[GlutenExternalAppendOnlyUnsafeRowArraySuite]  // 14 failures
+  enableSuite[GlutenExternalAppendOnlyUnsafeRowArraySuite]
   enableSuite[GlutenGlobalTempViewSuite]
   enableSuite[GlutenGlobalTempViewTestSuite]
   enableSuite[GlutenGroupedIteratorSuite]
@@ -696,18 +696,18 @@ class VeloxTestSettings extends BackendTestSettings {
   // TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite]  // 14 failures
   // TODO: 4.x enableSuite[GlutenRemoveRedundantSortsSuite]  // 1 failure
   enableSuite[GlutenRowToColumnConverterSuite]
-  // TODO: 4.x enableSuite[GlutenSQLExecutionSuite]  // 1 failure
+  enableSuite[GlutenSQLExecutionSuite]
   enableSuite[GlutenSQLFunctionSuite]
-  // TODO: 4.x enableSuite[GlutenSQLJsonProtocolSuite]  // 1 failure
-  // TODO: 4.x enableSuite[GlutenShufflePartitionsUtilSuite]  // 1 failure
+  enableSuite[GlutenSQLJsonProtocolSuite]
+  enableSuite[GlutenShufflePartitionsUtilSuite]
   // TODO: 4.x enableSuite[GlutenSimpleSQLViewSuite]  // 1 failure
   // TODO: 4.x enableSuite[GlutenSparkPlanSuite]  // 1 failure
   enableSuite[GlutenSparkPlannerSuite]
   enableSuite[GlutenSparkScriptTransformationSuite]
   enableSuite[GlutenSparkSqlParserSuite]
   enableSuite[GlutenUnsafeFixedWidthAggregationMapSuite]
   enableSuite[GlutenUnsafeKVExternalSorterSuite]
-  // TODO: 4.x enableSuite[GlutenUnsafeRowSerializerSuite]  // 1 failure
+  enableSuite[GlutenUnsafeRowSerializerSuite]
   // TODO: 4.x enableSuite[GlutenWholeStageCodegenSparkSubmitSuite]  // 1 failure
   // TODO: 4.x enableSuite[GlutenWholeStageCodegenSuite]  // 24 failures
   enableSuite[GlutenBroadcastExchangeSuite]
@@ -790,14 +790,7 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenSaveLoadSuite]
   enableSuite[GlutenTableScanSuite]
   // Generated suites for org.apache.spark.sql.sources
-  // TODO: 4.x enableSuite[GlutenBucketedReadWithHiveSupportSuite]  // 2 failures
-  // TODO: 4.x enableSuite[GlutenBucketedWriteWithHiveSupportSuite]  // 1 failure
-  // TODO: 4.x enableSuite[GlutenCommitFailureTestRelationSuite]  // 2 failures
   enableSuite[GlutenDataSourceAnalysisSuite]
-  // TODO: 4.x enableSuite[GlutenDisableUnnecessaryBucketedScanWithHiveSupportSuite]  // 2 failures
-  // TODO: 4.x enableSuite[GlutenJsonHadoopFsRelationSuite]  // 2 failures
-  // TODO: 4.x enableSuite[GlutenParquetHadoopFsRelationSuite]  // 2 failures
-  // TODO: 4.x enableSuite[GlutenSimpleTextHadoopFsRelationSuite]  // 2 failures
   // Generated suites for org.apache.spark.sql
   enableSuite[GlutenCacheManagerSuite]
   enableSuite[GlutenDataFrameShowSuite]
@@ -823,6 +816,7 @@ class VeloxTestSettings extends BackendTestSettings {
   // TODO: 4.x enableSuite[GlutenSetCommandSuite]  // 1 failure
   enableSuite[GlutenSparkSessionBuilderSuite]
   enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite]
+    .exclude("Tags set from session are prefixed with session UUID")
   enableSuite[GlutenTPCDSCollationQueryTestSuite]
   enableSuite[GlutenTPCDSModifiedPlanStabilitySuite]
   enableSuite[GlutenTPCDSModifiedPlanStabilityWithStatsSuite]

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSparkSessionJobTaggingAndCancellationSuite.scala

@@ -16,6 +16,71 @@
  */
 package org.apache.spark.sql
 
+import org.apache.spark.SparkContext
+import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
+import org.apache.spark.sql.execution.SQLExecution
+import org.apache.spark.util.ThreadUtils
+
+import java.util.concurrent.{Executors, Semaphore, TimeUnit}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
+
 class GlutenSparkSessionJobTaggingAndCancellationSuite
   extends SparkSessionJobTaggingAndCancellationSuite
-  with GlutenTestsTrait {}
+  with GlutenTestSetWithSystemPropertyTrait {
+
+  // Override with testGluten because the original test uses ExecutionContext.global whose
+  // ForkJoinPool reuses threads created before addTag("one"). InheritableThreadLocal (used by
+  // managedJobTags) only copies values at thread creation time, so reused threads see an empty
+  // tag map. This causes withSessionTagsApplied to not attach the user tag to the job, making
+  // cancelJobsWithTagWithFuture return empty. We fix this by creating a fresh thread pool AFTER
+  // addTag so that new threads inherit the InheritableThreadLocal values.
+  testGluten("Tags set from session are prefixed with session UUID") {
+    sc = new SparkContext("local[2]", "test")
+    val session = classic.SparkSession.builder().sparkContext(sc).getOrCreate()
+    import session.implicits._
+
+    val sem = new Semaphore(0)
+    sc.addSparkListener(new SparkListener {
+      override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+        sem.release()
+      }
+    })
+
+    session.addTag("one")
+
+    // Use a fresh thread pool created AFTER addTag so that new threads inherit
+    // InheritableThreadLocal values (managedJobTags, threadUuid).
+    val threadPool = Executors.newSingleThreadExecutor()
+    implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(threadPool)
+    try {
+      Future {
+        session.range(1, 10000).map { i => Thread.sleep(100); i }.count()
+      }
+
+      assert(sem.tryAcquire(1, 1, TimeUnit.MINUTES))
+      val activeJobsFuture =
+        session.sparkContext.cancelJobsWithTagWithFuture(
+          session.managedJobTags.get()("one"),
+          "reason")
+      val activeJob = ThreadUtils.awaitResult(activeJobsFuture, 60.seconds).head
+      val actualTags = activeJob.properties
+        .getProperty(SparkContext.SPARK_JOB_TAGS)
+        .split(SparkContext.SPARK_JOB_TAGS_SEP)
+      assert(
+        actualTags.toSet == Set(
+          session.sessionJobTag,
+          s"${session.sessionJobTag}-thread-${session.threadUuid.get()}-one",
+          SQLExecution.executionIdJobTag(
+            session,
+            activeJob.properties
+              .get(SQLExecution.EXECUTION_ROOT_ID_KEY)
+              .asInstanceOf[String]
+              .toLong)
+        ))
+    } finally {
+      threadPool.shutdownNow()
+    }
+  }
+}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenExternalAppendOnlyUnsafeRowArraySuite.scala

@@ -16,8 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenTestsTrait
+import org.apache.spark.sql.GlutenTestSetWithSystemPropertyTrait
 
 class GlutenExternalAppendOnlyUnsafeRowArraySuite
   extends ExternalAppendOnlyUnsafeRowArraySuite
-  with GlutenTestsTrait {}
+  with GlutenTestSetWithSystemPropertyTrait {}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenSQLExecutionSuite.scala

@@ -16,6 +16,6 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenTestsTrait
+import org.apache.spark.sql.GlutenTestSetWithSystemPropertyTrait
 
-class GlutenSQLExecutionSuite extends SQLExecutionSuite with GlutenTestsTrait {}
+class GlutenSQLExecutionSuite extends SQLExecutionSuite with GlutenTestSetWithSystemPropertyTrait {}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenSQLJsonProtocolSuite.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenTestsTrait
+import org.apache.spark.sql.GlutenTestSetWithSystemPropertyTrait
 
-class GlutenSQLJsonProtocolSuite extends SQLJsonProtocolSuite with GlutenTestsTrait {}
+class GlutenSQLJsonProtocolSuite
+  extends SQLJsonProtocolSuite
+  with GlutenTestSetWithSystemPropertyTrait {}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenShufflePartitionsUtilSuite.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenTestsTrait
+import org.apache.spark.sql.GlutenTestSetWithSystemPropertyTrait
 
-class GlutenShufflePartitionsUtilSuite extends ShufflePartitionsUtilSuite with GlutenTestsTrait {}
+class GlutenShufflePartitionsUtilSuite
+  extends ShufflePartitionsUtilSuite
+  with GlutenTestSetWithSystemPropertyTrait {}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenUnsafeRowSerializerSuite.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenTestsTrait
+import org.apache.spark.sql.GlutenTestSetWithSystemPropertyTrait
 
-class GlutenUnsafeRowSerializerSuite extends UnsafeRowSerializerSuite with GlutenTestsTrait {}
+class GlutenUnsafeRowSerializerSuite
+  extends UnsafeRowSerializerSuite
+  with GlutenTestSetWithSystemPropertyTrait {}