@@ -1618,13 +1618,17 @@ case class Cast(
   override def genCode(ctx: CodegenContext): ExprCode = {
     // If the cast does not change the structure, then we don't really need to cast anything.
     // We can return what the children return. Same thing should happen in the interpreted path.
-    if (DataType.equalsStructurally(child.dataType, dataType)) {
+    if (generatesChildCode) {
       child.genCode(ctx)
     } else {
       super.genCode(ctx)
     }
   }
 
+  /** Whether `genCode` generates the child's code as this cast's, the cast changing nothing. */
+  private[catalyst] def generatesChildCode: Boolean =
+    DataType.equalsStructurally(child.dataType, dataType)
+
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val eval = child.genCode(ctx)
     val nullSafeCast = nullSafeCastFunction(child.dataType, dataType, ctx)
 
@@ -26,6 +26,7 @@ import javax.tools.{Diagnostic, DiagnosticCollector, FileObject, ForwardingJavaF
 
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
+import scala.util.Try
 import scala.util.control.NonFatal
 
 import com.google.common.cache.{Cache, CacheBuilder}
@@ -326,14 +327,104 @@ object CodeCompiler extends Logging {
   private lazy val jitSkipsHugeMethods: Boolean =
     Utils.getVMOptionValue("DontCompileHugeMethods").contains("true")
 
-  /** Generated methods that run once per class or per partition, not per row. */
-  private val runOnceMethods = Set("<init>", "<clinit>", "init")
+  /**
+   * Generated methods that run once per class or per partition, not per row, so that one being
+   * interpreted costs nothing per row.
+   */
+  private[sql] val runOnceMethods = Set("<init>", "<clinit>", "init")
 
   private val hugeMethodWarned = new java.util.concurrent.atomic.AtomicBoolean(false)
 
   /** Lets a test see the first warning again. */
   private[catalyst] def resetHugeMethodWarning(): Unit = hugeMethodWarned.set(false)
 
+  /**
+   * A compile whose result its caller may discard ([[trial]]). What the compile would report of
+   * the code - its updates of the codegen metrics, which describe the code that runs, its "Code
+   * generated in" log line with the compile time, and its reports of methods past the JIT limit -
+   * is held here, in the order made, for the caller to make if it keeps the code.
+   *
+   * @param failureExpected whether a failure to compile is an answer the caller asks for, which the
+   *                        compile then logs at debug level rather than as an error.
+   */
+  private[sql] final class TrialCompile private[codegen] (val failureExpected: Boolean) {
+    private val methods = mutable.ArrayBuffer.empty[(String, Int)]
+    private val reports = mutable.ArrayBuffer.empty[() => Unit]
+
+    private[codegen] def hold(report: () => Unit): Unit = reports += report
+
+    private[codegen] def recordMethod(methodName: String, size: Int): Unit =
+      methods += ((methodName, size))
+
+    /**
+     * The bytecode size of every method the compile produced, by its name in the source, the
+     * largest where a name is in more than one class; empty where the compile was a hit in the
+     * compile cache, which produces nothing. Janino names a private method of an inner class with
+     * a `$` after its name in the source.
+     */
+    def methodSizes: Map[String, Int] =
+      methods.groupMapReduce(_._1.stripSuffix("$"))(_._2)(math.max)
+
+    /** Makes the reports and metric updates the compile held back, for the code kept. */
+    def report(): Unit = reports.foreach(_())
+  }
+
+  private val currentTrial = new ThreadLocal[TrialCompile]
+
+  /**
+   * Runs `body`, a compile whose result its caller may discard, holding back its codegen metric
+   * updates, its compile time log line and its reports of methods past the JIT limit until the
+   * caller keeps the code ([[TrialCompile.report]]); and, where `failureExpected`, logging its
+   * failure to compile at debug level, the failure being then the answer the caller is asking
+   * for. The source dump of `spark.sql.codegen.logLevel` is made for every compile, a trial's
+   * included. Whole-stage codegen compiles a stage this way to decide whether to split its
+   * expressions (`spark.sql.codegen.wholeStage.splitExpressions.methodLimit`). The trial is the
+   * calling thread's, and a backend that compiles on a thread of its own carries it there
+   * ([[withTrial]]), as the JDK backend does.
+   */
+  private[sql] def trial[T](failureExpected: Boolean)(body: => T): (Try[T], TrialCompile) = {
+    val trial = new TrialCompile(failureExpected)
+    (withTrial(trial)(Try(body)), trial)
+  }
+
+  /** The trial the calling thread compiles under, or null outside one. */
+  private[codegen] def activeTrial: TrialCompile = currentTrial.get
+
+  /**
+   * Makes `report`, a metric update or a log line about the code compiled - its compile time, or
+   * a method past the JIT limit - or, in a trial, holds it back until the caller keeps the code
+   * ([[TrialCompile.report]]), so that code compiled only to be discarded is neither counted in
+   * the metrics nor reported.
+   */
+  private[codegen] def reportOrHold(report: => Unit): Unit = Option(currentTrial.get) match {
+    case Some(trial) => trial.hold(() => report)
+    case None => report
+  }
+
+  /** Runs `body` on this thread under `trial`, which is null outside one. */
+  private[codegen] def withTrial[T](trial: TrialCompile)(body: => T): T = {
+    val previous = currentTrial.get
+    currentTrial.set(trial)
+    try body finally currentTrial.set(previous)
+  }
+
+  private def failureExpected: Boolean = {
+    val trial = currentTrial.get
+    trial != null && trial.failureExpected
+  }
+
+  /**
+   * Reports a failure to compile generated code, at debug level in a trial that expects one
+   * ([[trial]]).
+   */
+  private[codegen] def logCompileFailure(e: Throwable): Unit = {
+    if (failureExpected) {
+      logDebug("Failed to compile the generated Java code of a trial compile.", e)
+    } else {
+      logError("Failed to compile the generated Java code.", e)
+    }
+  }
+
   /**
    * Reports a generated method whose bytecode is past HotSpot's JIT limit
    * ([[CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT]]). The first time in the driver that such a
@@ -342,7 +433,8 @@ object CodeCompiler extends Logging {
    * once, and `spark.sql.codegen.hugeMethodLimit` does not already make the stage fall back.
    * Every other report is INFO: every one after that first warning, every one from a generated
    * class outside whole-stage codegen, where the setting does not apply, and every one on an
-   * executor, since the driver compiles each stage before its tasks do and has reported it.
+   * executor, since the driver compiles each stage before its tasks do and has reported it. In a
+   * trial, its caller holds the report back with the metric updates ([[reportOrHold]]).
    */
   private[catalyst] def logHugeMethod(className: String, methodName: String, size: Int): Unit = {
     val limit = CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT
@@ -378,16 +470,21 @@ object CodeCompiler extends Logging {
       classBytecodes: Iterable[(String, Array[Byte])]): ByteCodeStats = {
     val perClass = classBytecodes.map { case (_, classBytes) =>
       val classCodeSize = classBytes.length
-      CodegenMetrics.METRIC_GENERATED_CLASS_BYTECODE_SIZE.update(classCodeSize)
+      reportOrHold(CodegenMetrics.METRIC_GENERATED_CLASS_BYTECODE_SIZE.update(classCodeSize))
       try {
         val cf = new ClassFile(new ByteArrayInputStream(classBytes))
         val constPoolSize = cf.getConstantPoolSize
         val methodCodeSizes = cf.methodInfos.asScala.flatMap { method =>
           method.getAttributes.collect { case attr: CodeAttribute =>
             val byteCodeSize = attr.code.length
-            CodegenMetrics.METRIC_GENERATED_METHOD_BYTECODE_SIZE.update(byteCodeSize)
+            reportOrHold(CodegenMetrics.METRIC_GENERATED_METHOD_BYTECODE_SIZE.update(byteCodeSize))
+            Option(currentTrial.get).foreach(_.recordMethod(method.getName, byteCodeSize))
             if (byteCodeSize > CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT) {
-              logHugeMethod(cf.getThisClassName, method.getName, byteCodeSize)
+              // Bound first, so that a report held in a trial keeps the names, not the parsed
+              // class.
+              val className = cf.getThisClassName
+              val methodName = method.getName
+              reportOrHold(logHugeMethod(className, methodName, byteCodeSize))
             }
             byteCodeSize
           }
@@ -423,13 +520,76 @@ object CodeCompiler extends Logging {
    * the default conf instead of the calling session's.
    */
   private[codegen] def logGeneratedCodeOnFailure(code: CodeAndComment, maxLines: Int): Unit = {
+    if (failureExpected) {
+      return
+    }
     val formatted = s"\n${CodeFormatter.format(code, maxLines)}"
     if (Utils.isTesting) {
       logError(formatted)
     } else {
       logInfo(formatted)
     }
   }
+
+  /** What a span of generated Java source is, for [[forEachJavaSpan]]. */
+  private[codegen] sealed trait JavaSpan
+  private[codegen] object JavaSpan {
+    /** Code: whatever is neither a literal nor a comment. */
+    case object Code extends JavaSpan
+    /** A string or character literal, its quotes included. */
+    case object Literal extends JavaSpan
+    /** A `//` comment up to its line's end, or a block comment, its delimiters included. */
+    case object Comment extends JavaSpan
+  }
+
+  /**
+   * Calls `f` with each span of the generated Java `source` in turn: its kind, and where it
+   * starts and ends (exclusive). The spans cover the source. A literal's backslash escapes are
+   * honored, a `//` comment ends before its newline, and an unterminated literal or block comment
+   * runs to the end. The one scanner of generated Java: `JdkCodeCompiler.rewriteInnerClassRefs`
+   * rewrites the code spans, and `CodeGenerator.wholeStageSplitMarkersIn` reads the comments.
+   */
+  private[codegen] def forEachJavaSpan(source: String)(f: (JavaSpan, Int, Int) => Unit): Unit = {
+    val n = source.length
+    var i = 0
+    var codeStart = 0
+    def endCode(at: Int): Unit = if (at > codeStart) f(JavaSpan.Code, codeStart, at)
+    while (i < n) {
+      val c = source.charAt(i)
+      if (c == '"' || c == '\'') {
+        endCode(i)
+        val start = i
+        i += 1
+        var closed = false
+        while (i < n && !closed) {
+          val ch = source.charAt(i)
+          if (ch == '\\' && i + 1 < n) {
+            i += 2
+          } else {
+            i += 1
+            closed = ch == c
+          }
+        }
+        f(JavaSpan.Literal, start, i)
+        codeStart = i
+      } else if (c == '/' && i + 1 < n && source.charAt(i + 1) == '/') {
+        endCode(i)
+        val start = i
+        i = source.indexOf('\n', i) match { case -1 => n; case end => end }
+        f(JavaSpan.Comment, start, i)
+        codeStart = i
+      } else if (c == '/' && i + 1 < n && source.charAt(i + 1) == '*') {
+        endCode(i)
+        val start = i
+        i = source.indexOf("*/", i + 2) match { case -1 => n; case end => end + 2 }
+        f(JavaSpan.Comment, start, i)
+        codeStart = i
+      } else {
+        i += 1
+      }
+    }
+    endCode(n)
+  }
 }
 
 /**
@@ -472,11 +632,11 @@ object JaninoCodeCompiler extends CodeCompiler with Logging {
         CodeCompiler.computeByteCodeStats(evaluator.getBytecodes.asScala)
       } catch {
         case e: InternalCompilerException =>
-          logError("Failed to compile the generated Java code.", e)
+          CodeCompiler.logCompileFailure(e)
           CodeCompiler.logGeneratedCodeOnFailure(code, SQLConf.get.loggingMaxLinesForCodegen)
           throw QueryExecutionErrors.internalCompilerError(e)
         case e: CompileException =>
-          logError("Failed to compile the generated Java code.", e)
+          CodeCompiler.logCompileFailure(e)
           CodeCompiler.logGeneratedCodeOnFailure(code, SQLConf.get.loggingMaxLinesForCodegen)
           throw QueryExecutionErrors.compilerError(e)
       }
@@ -697,68 +857,19 @@ object JdkCodeCompiler extends CodeCompiler with Logging {
    * The rewrite is applied only to actual code spans: string literals, char
    * literals, and `//` / block comments are copied verbatim so that a `$Upper`
    * sequence inside generated string data (e.g. a column name or error message)
-   * is never corrupted.
+   * is never corrupted. The spans come from `CodeCompiler.forEachJavaSpan`.
    */
   private[codegen] def rewriteInnerClassRefs(body: String, classLoader: ClassLoader): String = {
     val out = new java.lang.StringBuilder(body.length + 16)
-    val code = new java.lang.StringBuilder()
-    val n = body.length
     // A token's rewritten form is stable for a given classloader; memoize within
     // this call so repeated type references resolve at most once.
     val memo = mutable.HashMap.empty[String, String]
-
-    def flushCode(): Unit = {
-      if (code.length > 0) {
-        out.append(rewriteCodeSpan(code.toString, classLoader, memo))
-        code.setLength(0)
-      }
-    }
-
-    // Copy a quoted literal (string or char) verbatim, honoring backslash escapes.
-    def copyQuoted(start: Int, quote: Char): Int = {
-      out.append(quote)
-      var j = start + 1
-      var closed = false
-      while (j < n && !closed) {
-        val ch = body.charAt(j)
-        if (ch == '\\' && j + 1 < n) {
-          out.append(ch).append(body.charAt(j + 1))
-          j += 2
-        } else {
-          out.append(ch)
-          j += 1
-          if (ch == quote) closed = true
-        }
-      }
-      j
-    }
-
-    var i = 0
-    while (i < n) {
-      val c = body.charAt(i)
-      if (c == '"' || c == '\'') {
-        flushCode()
-        i = copyQuoted(i, c)
-      } else if (c == '/' && i + 1 < n && body.charAt(i + 1) == '/') {
-        flushCode()
-        while (i < n && body.charAt(i) != '\n') { out.append(body.charAt(i)); i += 1 }
-      } else if (c == '/' && i + 1 < n && body.charAt(i + 1) == '*') {
-        flushCode()
-        out.append("/*")
-        i += 2
-        while (i < n && !(body.charAt(i) == '*' && i + 1 < n && body.charAt(i + 1) == '/')) {
-          out.append(body.charAt(i)); i += 1
-        }
-        // The scan exits either at the `*/` terminator (then i + 1 < n holds by the
-        // loop condition) or at end-of-body for an unterminated comment, whose
-        // characters the loop already copied verbatim.
-        if (i + 1 < n) { out.append("*/"); i += 2 }
-      } else {
-        code.append(c)
-        i += 1
-      }
+    CodeCompiler.forEachJavaSpan(body) {
+      case (CodeCompiler.JavaSpan.Code, start, end) =>
+        out.append(rewriteCodeSpan(body.substring(start, end), classLoader, memo))
+      case (_, start, end) =>
+        out.append(body, start, end)
     }
-    flushCode()
     out.toString
   }
 
@@ -1316,9 +1427,12 @@ object JdkCodeCompiler extends CodeCompiler with Logging {
     // not carry a cause, or downstream resolution can fail in non-local deployments.
     val parentLoader = new ParentClassLoader(resolveLoader)
 
+    // The compile reports from the worker, so a trial compile's marker goes there with it.
+    val trial = CodeCompiler.activeTrial
     val future = compileExecutor.submit(new Callable[(GeneratedClass, ByteCodeStats)] {
-      override def call(): (GeneratedClass, ByteCodeStats) =
+      override def call(): (GeneratedClass, ByteCodeStats) = CodeCompiler.withTrial(trial) {
         doCompile(code, source, resolveLoader, parentLoader, failureLogMaxLines)
+      }
     })
     try {
       // Await uninterruptibly: the result is cached and the worker must finish its
@@ -1370,7 +1484,7 @@ object JdkCodeCompiler extends CodeCompiler with Logging {
       task.call().booleanValue()
     } catch {
       case NonFatal(e) =>
-        logError("Failed to compile the generated Java code.", e)
+        CodeCompiler.logCompileFailure(e)
         logSourceOnFailure()
         throw QueryExecutionErrors.internalCompilerError(
           new InternalCompilerException(e.getMessage, e))
@@ -1389,7 +1503,7 @@ object JdkCodeCompiler extends CodeCompiler with Logging {
         case parts => parts.mkString("\n")
       }
       val ex = new CompileException(message, null)
-      logError("Failed to compile the generated Java code.", ex)
+      CodeCompiler.logCompileFailure(ex)
       logSourceOnFailure()
       throw QueryExecutionErrors.compilerError(ex)
     }