[SPARK-56353][BUILD][TESTS] Generate Java-based test JARs dynamically instead of storing pre-built binaries

### What changes were proposed in this pull request?
This PR is a part of SPARK-56352 for Java-based test JARs, replacing pre-built test JAR files containing Java classes with dynamic compilation at test time, removing 7 binary JAR files from the repository.

Changes:
- Add `TestUtils.createJarWithJavaSources()` in `SparkTestUtils.scala` that compiles Java source code with `javac` and packages the resulting classes into a JAR.
- Externalize Java source files for Hive UDFs/UDAFs/UDTFs to `src/test/resources/` (9 files), loaded at test time via  `getContextClassLoader.getResource()`. These source files were reverse-engineered from the class files in the pre-built JARs, as no original source code existed in the repository.
- Update test suites to use dynamically generated JARs instead of pre-built ones.
- Remove deleted JAR entries from `dev/test-jars.txt`.

JARs removed:
- `core/src/test/resources/TestUDTF.jar`
- `sql/core/src/test/resources/SPARK-33084.jar`
- `sql/hive-thriftserver/src/test/resources/TestUDTF.jar`
- `sql/hive/src/test/noclasspath/hive-test-udfs.jar`
- `sql/hive/src/test/resources/SPARK-21101-1.0.jar`
- `sql/hive/src/test/resources/TestUDTF.jar`
- `sql/hive/src/test/resources/data/files/TestSerDe.jar`

### Why are the changes needed?
As noted in the PR discussion [here](#50378 (review)):
> the ultimate goal is to refactor the tests to automatically build the jars instead of using pre-built ones

This PR achieves that goal for all Java-based test JARs. By generating JARs dynamically at test time, no binary artifacts need to be stored in the source tree, and the release-time workaround becomes unnecessary for these files.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
All affected test suites pass:
- SparkContextSuite, SparkSubmitSuite, TaskSetManagerSuite (core)
- SQLQuerySuite (sql/core)
- HiveUDFDynamicLoadSuite, HiveDDLSuite, HiveQuerySuite, HiveUDFSuite, SQLQuerySuite (sql/hive)
- CliSuite, HiveThriftServer2Suites (sql/hive-thriftserver)

No tests were added or removed. Existing tests now compile Java sources at
runtime instead of loading pre-built JARs.

### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Opus 4.6

Closes #55192 from sarutak/remove-test-jars-ab.

Authored-by: Kousuke Saruta <sarutak@amazon.co.jp>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: sarutak

common/utils/src/main/scala/org/apache/spark/util/SparkTestUtils.scala

@@ -17,10 +17,11 @@
 
 package org.apache.spark.util
 
-import java.io.File
+import java.io.{File, FileInputStream, FileOutputStream}
 import java.net.{URI, URL}
 import java.nio.file.Files
 import java.util.Arrays
+import java.util.jar.{JarEntry, JarOutputStream}
 import javax.tools.{JavaFileObject, SimpleJavaFileObject, ToolProvider}
 
 import scala.jdk.CollectionConverters._
@@ -99,6 +100,69 @@ private[spark] trait SparkTestUtils {
     createCompiledClass(className, destDir, sourceFile, classpathUrls)
   }
 
+  /**
+   * Compile Java source code and package the resulting class files into a JAR.
+   * Supports classes with package declarations - the JAR will contain the proper
+   * directory structure (e.g., org/apache/spark/Foo.class).
+   *
+   * @param sources map of fully-qualified class name to Java source code
+   * @param jarFile the JAR file to create
+   * @param classpathUrls additional classpath URLs needed for compilation
+   * @return URL of the created JAR file
+   */
+  def createJarWithJavaSources(
+      sources: Map[String, String],
+      jarFile: File,
+      classpathUrls: Seq[URL] = Seq.empty): URL = {
+    val compiler = ToolProvider.getSystemJavaCompiler
+    val classDir = Files.createTempDirectory("spark-test-classes").toFile
+
+    val sourceFiles = sources.map { case (name, code) =>
+      new JavaSourceFromString(name, code)
+    }.toList
+
+    val options = Seq("-d", classDir.getAbsolutePath) ++ (
+      if (classpathUrls.nonEmpty) {
+        Seq("-classpath", classpathUrls.map(_.getFile).mkString(File.pathSeparator))
+      } else Seq.empty
+    )
+
+    val success = compiler.getTask(
+      null, null, null, options.asJava, null,
+      java.util.Arrays.asList(sourceFiles: _*)).call()
+    assert(success, s"Compilation failed for: ${sources.keys.mkString(", ")}")
+
+    // Collect all .class files under classDir
+    val classFiles = listFilesRecursively(classDir).filter(_.getName.endsWith(".class"))
+
+    val jarStream = new JarOutputStream(new FileOutputStream(jarFile))
+    try {
+      for (classFile <- classFiles) {
+        val entryName = classDir.toPath.relativize(classFile.toPath).toString.replace('\\', '/')
+        jarStream.putNextEntry(new JarEntry(entryName))
+        val in = new FileInputStream(classFile)
+        try { in.transferTo(jarStream) } finally { in.close() }
+      }
+    } finally {
+      jarStream.close()
+    }
+
+    SparkFileUtils.deleteRecursively(classDir)
+    jarFile.toURI.toURL
+  }
+
+
+  private def listFilesRecursively(dir: File): Seq[File] = {
+    val children = dir.listFiles()
+    if (children == null) {
+      Seq.empty
+    } else {
+      children.flatMap { f =>
+        if (f.isDirectory) { listFilesRecursively(f) } else { Seq(f) }
+      }.toSeq
+    }
+  }
+
 }
 
 private[spark] object SparkTestUtils extends SparkTestUtils

core/src/test/resources/TestUDTF.jar

@@ -244,23 +244,22 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
   }
 
   test("add and list jar files") {
-    val testJar = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar")
-    assume(testJar != null)
+    val testJar = TestUtils.createJarWithClasses(Seq("SparkContextSuite_AddJar"))
     try {
       sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
       sc.addJar(testJar.toString)
-      assert(sc.listJars().count(_.contains("TestUDTF.jar")) == 1)
+      assert(sc.listJars().count(_.contains("testJar")) == 1)
     } finally {
       sc.stop()
     }
   }
 
   test("add FS jar files not exists") {
     try {
-      val jarPath = "hdfs:///no/path/to/TestUDTF.jar"
+      val jarPath = "hdfs:///no/path/to/nonexistent.jar"
       sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
       sc.addJar(jarPath)
-      assert(sc.listJars().forall(!_.contains("TestUDTF.jar")))
+      assert(sc.listJars().forall(!_.contains("nonexistent.jar")))
     } finally {
       sc.stop()
     }
@@ -402,8 +401,7 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
       case "non-local-mode" => "local-cluster[1,1,1024]"
     }
     test(s"$method can be called twice with same file in $schedulingMode (SPARK-16787)") {
-      val testJar = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar")
-      assume(testJar != null)
+      val testJar = TestUtils.createJarWithClasses(Seq("SparkContextSuite_SPARK16787"))
       sc = new SparkContext(master, "test")
       val jarPath = testJar.toString
       method match {

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -109,6 +109,10 @@ class SparkSubmitSuite
   private val emptyIvySettings = File.createTempFile("ivy", ".xml")
   Files.writeString(emptyIvySettings.toPath, "<ivysettings />")
 
+  private val testJarPath: String = {
+    val url = TestUtils.createJarWithClasses(Seq("SparkSubmitSuite_Dummy"))
+    new File(url.toURI).getAbsolutePath
+  }
   private val submit = new SparkSubmit()
 
   override def beforeEach(): Unit = {
@@ -502,8 +506,7 @@ class SparkSubmitSuite
 
   test("SPARK-47495: Not to add primary resource to jars again" +
     " in k8s client mode & driver runs inside a POD") {
-    val testJar = "src/test/resources/TestUDTF.jar"
-    assume(new File(testJar).exists)
+    val testJar = testJarPath
     val clArgs = Seq(
       "--deploy-mode", "client",
       "--proxy-user", "test.user",
@@ -518,12 +521,11 @@ class SparkSubmitSuite
     val appArgs = new SparkSubmitArguments(clArgs)
     val (_, _, sparkConf, _) = submit.prepareSubmitEnvironment(appArgs)
     sparkConf.get("spark.jars").contains("jarToIgnore") shouldBe false
-    sparkConf.get("spark.jars").contains("TestUDTF") shouldBe true
+    sparkConf.get("spark.jars").contains("testJar") shouldBe true
   }
 
   test("SPARK-33782: handles k8s files download to current directory") {
-    val testJar = "src/test/resources/TestUDTF.jar"
-    assume(new File(testJar).exists)
+    val testJar = testJarPath
     val clArgs = Seq(
       "--deploy-mode", "client",
       "--proxy-user", "test.user",
@@ -548,21 +550,22 @@ class SparkSubmitSuite
     conf.get("spark.kubernetes.namespace") should be ("spark")
     conf.get("spark.kubernetes.driver.container.image") should be ("bar")
 
+    val testJarName = new File(testJar).getName
     Files.exists(Paths.get("test_metrics_config.properties")) should be (true)
     Files.exists(Paths.get("test_metrics_system.properties")) should be (true)
     Files.exists(Paths.get("log4j2.properties")) should be (true)
-    Files.exists(Paths.get("TestUDTF.jar")) should be (true)
+    Files.exists(Paths.get(testJarName)) should be (true)
     Files.delete(Paths.get("test_metrics_config.properties"))
     Files.delete(Paths.get("test_metrics_system.properties"))
     Files.delete(Paths.get("log4j2.properties"))
-    Files.delete(Paths.get("TestUDTF.jar"))
+    Files.delete(Paths.get(testJarName))
   }
 
   test("SPARK-47475: Avoid jars download if scheme matches " +
     "spark.kubernetes.jars.avoidDownloadSchemes " +
     "in k8s client mode & driver runs inside a POD") {
-    val testJar = "src/test/resources/TestUDTF.jar"
-    assume(new File(testJar).exists)
+    val testJar = testJarPath
+    val testJarName = new File(testJar).getName
     val hadoopConf = new Configuration()
     updateConfWithFakeS3Fs(hadoopConf)
     withTempDir { tmpDir =>
@@ -588,17 +591,17 @@ class SparkSubmitSuite
       val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs, Some(hadoopConf))
       conf.get("spark.master") should be("k8s://https://host:port")
       conf.get("spark.jars").contains(remoteJarFile) shouldBe true
-      conf.get("spark.jars").contains("TestUDTF") shouldBe true
+      conf.get("spark.jars").contains(testJarName) shouldBe true
 
       Files.exists(Paths.get("test_metrics_config.properties")) should be(true)
       Files.exists(Paths.get("test_metrics_system.properties")) should be(true)
       Files.exists(Paths.get("log4j2.properties")) should be(true)
-      Files.exists(Paths.get("TestUDTF.jar")) should be(true)
+      Files.exists(Paths.get(testJarName)) should be(true)
       Files.exists(Paths.get(notToDownload.getName)) should be(false)
       Files.delete(Paths.get("test_metrics_config.properties"))
       Files.delete(Paths.get("test_metrics_system.properties"))
       Files.delete(Paths.get("log4j2.properties"))
-      Files.delete(Paths.get("TestUDTF.jar"))
+      Files.delete(Paths.get(testJarName))
     }
   }
 
@@ -1932,7 +1935,7 @@ class SparkSubmitSuite
       TestUtils.createJar(Seq(text1), zipFile1)
       val testFile = "test_metrics_config.properties"
       val testPyFile = "test_metrics_system.properties"
-      val testJar = "TestUDTF.jar"
+      val testJar = new File(testJarPath).getName
       val clArgs = Seq(
         "--deploy-mode", "client",
         "--proxy-user", "test.user",
@@ -1945,7 +1948,7 @@ class SparkSubmitSuite
         "--conf", "spark.kubernetes.submitInDriver=true",
         "--files", s"src/test/resources/$testFile",
         "--py-files", s"src/test/resources/$testPyFile",
-        "--jars", s"src/test/resources/$testJar",
+        "--jars", testJarPath,
         "--archives", s"${zipFile1.getAbsolutePath}#test_archives",
         "/home/thejar.jar",
         "arg1")

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -33,7 +33,7 @@ class ClassLoaderIsolationSuite extends SparkFunSuite with LocalSparkContext  {
     .take(2)
     .mkString(".")
 
-  private val jarURL1 = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar")
+  private val jarURL1 = TestUtils.createJarWithClasses(Seq("ClassLoaderIsolation_Dummy"))
   private lazy val jar1 = jarURL1.toString
 
   // package com.example
@@ -51,7 +51,6 @@ class ClassLoaderIsolationSuite extends SparkFunSuite with LocalSparkContext  {
   private lazy val jar3 = jarURL3.toString
 
   test("Executor classloader isolation with JobArtifactSet") {
-    assume(jarURL1 != null)
     assume(jarURL2 != null)
     assume(jarURL3 != null)
 

core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala

@@ -1573,8 +1573,7 @@ class TaskSetManagerSuite
   }
 
   test("SPARK-21563 context's added jars shouldn't change mid-TaskSet") {
-    val jarPath = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar")
-    assume(jarPath != null)
+    val jarPath = TestUtils.createJarWithClasses(Seq("TaskSetManagerSuite_SPARK21563"))
     sc = new SparkContext("local", "test")
     val addedJarsPreTaskSet = Map[String, Long](sc.allAddedJars.toSeq: _*)
     assert(addedJarsPreTaskSet.size === 0)

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -1,17 +1,10 @@
 core/src/test/resources/TestHelloV2_2.13.jar
 core/src/test/resources/TestHelloV3_2.13.jar
-core/src/test/resources/TestUDTF.jar
 data/artifact-tests/junitLargeJar.jar
 data/artifact-tests/smallJar.jar
 sql/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar
 sql/connect/client/jvm/src/test/resources/udf2.13.jar
 sql/connect/common/src/test/resources/artifact-tests/junitLargeJar.jar
 sql/connect/common/src/test/resources/artifact-tests/smallJar.jar
-sql/core/src/test/resources/SPARK-33084.jar
 sql/core/src/test/resources/artifact-tests/udf_noA.jar
-sql/hive-thriftserver/src/test/resources/TestUDTF.jar
-sql/hive/src/test/noclasspath/hive-test-udfs.jar
-sql/hive/src/test/resources/SPARK-21101-1.0.jar
-sql/hive/src/test/resources/TestUDTF.jar
-sql/hive/src/test/resources/data/files/TestSerDe.jar
 sql/hive/src/test/resources/regression-test-SPARK-8489/test-2.13.jar

dev/test-jars.txt

@@ -0,0 +1,64 @@
+/*
+ * 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.examples.sql;
+
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.expressions.MutableAggregationBuffer;
+import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
+import org.apache.spark.sql.types.*;
+
+public class Spark33084 extends UserDefinedAggregateFunction {
+  public StructType inputSchema() {
+    return new StructType().add("inputColumn", DataTypes.LongType);
+  }
+
+  public StructType bufferSchema() {
+    return new StructType()
+        .add("sum", DataTypes.LongType)
+        .add("count", DataTypes.LongType);
+  }
+
+  public DataType dataType() {
+    return DataTypes.DoubleType;
+  }
+
+  public boolean deterministic() {
+    return true;
+  }
+
+  public void initialize(MutableAggregationBuffer buffer) {
+    buffer.update(0, 0L);
+    buffer.update(1, 0L);
+  }
+
+  public void update(MutableAggregationBuffer buffer, Row input) {
+    if (!input.isNullAt(0)) {
+      buffer.update(0, buffer.getLong(0) + input.getLong(0));
+      buffer.update(1, buffer.getLong(1) + 1L);
+    }
+  }
+
+  public void merge(MutableAggregationBuffer buffer1, Row buffer2) {
+    buffer1.update(0, buffer1.getLong(0) + buffer2.getLong(0));
+    buffer1.update(1, buffer1.getLong(1) + buffer2.getLong(1));
+  }
+
+  public Object evaluate(Row buffer) {
+    return (double) buffer.getLong(0) / (double) buffer.getLong(1);
+  }
+}

sql/core/src/test/resources/SPARK-33084.jar

@@ -18,7 +18,10 @@
 package org.apache.spark.sql
 
 import java.io.File
+import java.lang.management.ManagementFactory
 import java.net.{MalformedURLException, URI}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Paths}
 import java.sql.{Date, Timestamp}
 import java.time.{Duration, Period}
 import java.util.Locale
@@ -56,7 +59,7 @@ import org.apache.spark.sql.test.SQLTestData._
 import org.apache.spark.sql.types._
 import org.apache.spark.tags.ExtendedSQLTest
 import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
-import org.apache.spark.util.{ResetSystemProperties, Utils}
+import org.apache.spark.util.{ResetSystemProperties, SparkTestUtils, Utils}
 
 @ExtendedSQLTest
 class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper
@@ -3865,19 +3868,25 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
   }
 
   test("SPARK-33084: Add jar support Ivy URI in SQL -- jar contains udf class") {
-    val jarPath = Thread.currentThread().getContextClassLoader
-      .getResource("SPARK-33084.jar")
-    assume(jarPath != null)
     val sumFuncClass = "org.apache.spark.examples.sql.Spark33084"
+    val resourceName = "SPARK-33084/Spark33084.java"
+    val sourceUrl = Thread.currentThread().getContextClassLoader
+      .getResource(resourceName)
+    assert(sourceUrl != null, s"Resource not found: $resourceName")
+    val source = Map(sumFuncClass ->
+      new String(Files.readAllBytes(Paths.get(sourceUrl.toURI)), StandardCharsets.UTF_8))
+    val classpath = ManagementFactory.getRuntimeMXBean.getClassPath
+      .split(File.pathSeparator).map(p => new File(p).toURI.toURL).toSeq
+    val jarFile = new File(Utils.createTempDir(), "SPARK-33084.jar")
+    SparkTestUtils.createJarWithJavaSources(source, jarFile, classpath)
     val functionName = "test_udf"
     withTempDir { dir =>
       System.setProperty("ivy.home", dir.getAbsolutePath)
-      val sourceJar = new File(jarPath.getFile)
       val targetCacheJarDir = new File(dir.getAbsolutePath +
         "/local/org.apache.spark/SPARK-33084/1.0/jars/")
       targetCacheJarDir.mkdir()
       // copy jar to local cache
-      Utils.copyFileToDirectory(sourceJar, targetCacheJarDir)
+      Utils.copyFileToDirectory(jarFile, targetCacheJarDir)
       withTempView("v1") {
         withUserDefinedFunction(
           s"default.$functionName" -> false,