cloud-fan commented on code in PR #56430: URL: https://github.com/apache/spark/pull/56430#discussion_r3708708488
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeCompiler.scala: ########## @@ -0,0 +1,1279 @@ +/* + * 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.catalyst.expressions.codegen + +import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, IOException, StringWriter} +import java.net.{JarURLConnection, URI, URL} +import java.util.Locale +import java.util.concurrent.{Callable, ExecutionException, ExecutorService} +import javax.tools.{Diagnostic, DiagnosticCollector, FileObject, ForwardingJavaFileManager, JavaCompiler, JavaFileManager, JavaFileObject, SimpleJavaFileObject, StandardJavaFileManager, StandardLocation, ToolProvider} + +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +import com.google.common.cache.{Cache, CacheBuilder} +import com.google.common.util.concurrent.Uninterruptibles +import org.codehaus.commons.compiler.{CompileException, InternalCompilerException} +import org.codehaus.janino.ClassBodyEvaluator +import org.codehaus.janino.util.ClassFile +import org.codehaus.janino.util.ClassFile.CodeAttribute + +import org.apache.spark.{JobArtifactSet, SparkEnv, TaskContext, TaskKilledException} +import org.apache.spark.executor.{ExecutorClassLoader, InputMetrics} +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.metrics.source.CodegenMetrics +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Expression, UnsafeArrayData, UnsafeMapData, UnsafeRow} +import org.apache.spark.sql.catalyst.util.{ArrayData, CollationAwareUTF8String, CollationFactory, CollationSupport, MapData} +import org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.Decimal +import org.apache.spark.unsafe.Platform +import org.apache.spark.unsafe.types.{BinaryView, CalendarInterval, TimestampNanosVal, UTF8String, VariantVal} +import org.apache.spark.util.{ParentClassLoader, ThreadUtils, Utils} + +/** + * Backend used to compile generator-produced Java source into a [[GeneratedClass]]. + * + * Two implementations are provided: + * - [[JaninoCodeCompiler]]: the default, uses Janino's `ClassBodyEvaluator`. Very fast. + * - [[JdkCodeCompiler]]: uses `javax.tools.JavaCompiler` from the JDK. Slower (~5x for + * large generated units, 30-300x for small ones), but maintained on the JDK + * release cadence and not subject to Janino's unmaintained-upstream risk. + * + * The backend is selected at compile time via [[SQLConf.CODEGEN_COMPILER]]. + */ +trait CodeCompiler { + /** Backend name as used in `spark.sql.codegen.compiler`. */ + def name: String + + /** + * Compile a generator-produced class body into an instance of the + * [[GeneratedClass]] subclass it defines. + * + * @return the instantiated generated class along with bytecode statistics. + */ + def compile(code: CodeAndComment): (GeneratedClass, ByteCodeStats) +} + +object CodeCompiler extends Logging { + + // Emit log messages under CodeGenerator's logger name: operators and tests + // (SPARK-25113 / SPARK-51527) subscribe to that exact logger for codegen + // compilation events, and the backends are implementation details of + // `CodeGenerator.compile`, so their logs belong under its name. + override protected def logName: String = classOf[CodeGenerator[_, _]].getName + + val JANINO: String = "janino" + val JDK: String = "jdk" + + /** + * Fully-qualified imports made available to generated code by both backends. + * + * For Janino these are passed to `ClassBodyEvaluator.setDefaultImports`. + * For the JDK backend they are rendered into `import` statements inside the + * synthesized compilation unit. + * + * This is the single shared list - anything added here automatically applies to + * both backends. It intentionally excludes + * `org.apache.spark.sql.catalyst.expressions.codegen.GeneratedClass` to avoid + * a name collision with the generated subclass `GeneratedClass`; the extends + * clause uses the fully-qualified name instead. + * + * When adding an entry, keep its SIMPLE name distinct from any `import` line a + * generator emits at the top of a class body (currently only GenerateColumnAccessor + * does this): javac rejects two single-type imports sharing a simple name + * (JLS 7.5.1) while Janino resolves them leniently, so a collision would fail only + * under the JDK backend. + */ + val DefaultImports: Seq[String] = Seq( + classOf[Platform].getName, + classOf[InternalRow].getName, + classOf[UnsafeRow].getName, + classOf[BinaryView].getName, + classOf[UTF8String].getName, + classOf[Decimal].getName, + classOf[CalendarInterval].getName, + classOf[TimestampNanosVal].getName, + classOf[VariantVal].getName, + classOf[ArrayData].getName, + classOf[UnsafeArrayData].getName, + classOf[MapData].getName, + classOf[UnsafeMapData].getName, + classOf[Expression].getName, + classOf[TaskContext].getName, + classOf[TaskKilledException].getName, + classOf[InputMetrics].getName, + classOf[CollationAwareUTF8String].getName, + classOf[CollationFactory].getName, + classOf[CollationSupport].getName, + QueryExecutionErrors.getClass.getName.stripSuffix("$") + ) + + /** + * FQN of the generated class. Must NOT be under the `codegen` package or Janino + * fails with `java.lang.InstantiationException`. The same name is used for both + * backends so generated source, logs, and diagnostics name the same class + * whichever backend compiles it. (Compiled results are NOT shared across + * backends: the compile cache key includes the backend.) + */ + val GeneratedClassName: String = + "org.apache.spark.sql.catalyst.expressions.GeneratedClass" + + def active(): CodeCompiler = active(null) + + /** + * Resolve the active backend for the given generated unit. + * + * The configured backend ([[SQLConf.CODEGEN_COMPILER]]) governs ordinary codegen. The + * exception is codegen the JDK compiler is fundamentally *incapable* of compiling - not + * merely slower at - which is always routed to Janino regardless of the configured Review Comment: `merely slower at` is missing its object; please use `merely slower at compiling`. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeCompiler.scala: ########## @@ -0,0 +1,1279 @@ +/* + * 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.catalyst.expressions.codegen + +import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, IOException, StringWriter} +import java.net.{JarURLConnection, URI, URL} +import java.util.Locale +import java.util.concurrent.{Callable, ExecutionException, ExecutorService} +import javax.tools.{Diagnostic, DiagnosticCollector, FileObject, ForwardingJavaFileManager, JavaCompiler, JavaFileManager, JavaFileObject, SimpleJavaFileObject, StandardJavaFileManager, StandardLocation, ToolProvider} + +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +import com.google.common.cache.{Cache, CacheBuilder} +import com.google.common.util.concurrent.Uninterruptibles +import org.codehaus.commons.compiler.{CompileException, InternalCompilerException} +import org.codehaus.janino.ClassBodyEvaluator +import org.codehaus.janino.util.ClassFile +import org.codehaus.janino.util.ClassFile.CodeAttribute + +import org.apache.spark.{JobArtifactSet, SparkEnv, TaskContext, TaskKilledException} +import org.apache.spark.executor.{ExecutorClassLoader, InputMetrics} +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.metrics.source.CodegenMetrics +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Expression, UnsafeArrayData, UnsafeMapData, UnsafeRow} +import org.apache.spark.sql.catalyst.util.{ArrayData, CollationAwareUTF8String, CollationFactory, CollationSupport, MapData} +import org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.Decimal +import org.apache.spark.unsafe.Platform +import org.apache.spark.unsafe.types.{BinaryView, CalendarInterval, TimestampNanosVal, UTF8String, VariantVal} +import org.apache.spark.util.{ParentClassLoader, ThreadUtils, Utils} + +/** + * Backend used to compile generator-produced Java source into a [[GeneratedClass]]. + * + * Two implementations are provided: + * - [[JaninoCodeCompiler]]: the default, uses Janino's `ClassBodyEvaluator`. Very fast. + * - [[JdkCodeCompiler]]: uses `javax.tools.JavaCompiler` from the JDK. Slower (~5x for + * large generated units, 30-300x for small ones), but maintained on the JDK + * release cadence and not subject to Janino's unmaintained-upstream risk. + * + * The backend is selected at compile time via [[SQLConf.CODEGEN_COMPILER]]. + */ +trait CodeCompiler { + /** Backend name as used in `spark.sql.codegen.compiler`. */ + def name: String + + /** + * Compile a generator-produced class body into an instance of the + * [[GeneratedClass]] subclass it defines. + * + * @return the instantiated generated class along with bytecode statistics. + */ + def compile(code: CodeAndComment): (GeneratedClass, ByteCodeStats) +} + +object CodeCompiler extends Logging { + + // Emit log messages under CodeGenerator's logger name: operators and tests + // (SPARK-25113 / SPARK-51527) subscribe to that exact logger for codegen + // compilation events, and the backends are implementation details of + // `CodeGenerator.compile`, so their logs belong under its name. + override protected def logName: String = classOf[CodeGenerator[_, _]].getName + + val JANINO: String = "janino" + val JDK: String = "jdk" + + /** + * Fully-qualified imports made available to generated code by both backends. + * + * For Janino these are passed to `ClassBodyEvaluator.setDefaultImports`. + * For the JDK backend they are rendered into `import` statements inside the + * synthesized compilation unit. + * + * This is the single shared list - anything added here automatically applies to + * both backends. It intentionally excludes + * `org.apache.spark.sql.catalyst.expressions.codegen.GeneratedClass` to avoid + * a name collision with the generated subclass `GeneratedClass`; the extends + * clause uses the fully-qualified name instead. + * + * When adding an entry, keep its SIMPLE name distinct from any `import` line a + * generator emits at the top of a class body (currently only GenerateColumnAccessor + * does this): javac rejects two single-type imports sharing a simple name + * (JLS 7.5.1) while Janino resolves them leniently, so a collision would fail only + * under the JDK backend. + */ + val DefaultImports: Seq[String] = Seq( + classOf[Platform].getName, + classOf[InternalRow].getName, + classOf[UnsafeRow].getName, + classOf[BinaryView].getName, + classOf[UTF8String].getName, + classOf[Decimal].getName, + classOf[CalendarInterval].getName, + classOf[TimestampNanosVal].getName, + classOf[VariantVal].getName, + classOf[ArrayData].getName, + classOf[UnsafeArrayData].getName, + classOf[MapData].getName, + classOf[UnsafeMapData].getName, + classOf[Expression].getName, + classOf[TaskContext].getName, + classOf[TaskKilledException].getName, + classOf[InputMetrics].getName, + classOf[CollationAwareUTF8String].getName, + classOf[CollationFactory].getName, + classOf[CollationSupport].getName, + QueryExecutionErrors.getClass.getName.stripSuffix("$") + ) + + /** + * FQN of the generated class. Must NOT be under the `codegen` package or Janino + * fails with `java.lang.InstantiationException`. The same name is used for both + * backends so generated source, logs, and diagnostics name the same class + * whichever backend compiles it. (Compiled results are NOT shared across + * backends: the compile cache key includes the backend.) + */ + val GeneratedClassName: String = + "org.apache.spark.sql.catalyst.expressions.GeneratedClass" + + def active(): CodeCompiler = active(null) + + /** + * Resolve the active backend for the given generated unit. + * + * The configured backend ([[SQLConf.CODEGEN_COMPILER]]) governs ordinary codegen. The + * exception is codegen the JDK compiler is fundamentally *incapable* of compiling - not + * merely slower at - which is always routed to Janino regardless of the configured + * backend. This is deterministic routing decided up front from the execution context and + * the generated source; it is never a fallback after a failed compile. Two such cases, + * both classes the JDK compiler cannot name that Janino's lenient loader/lexer accepts: Review Comment: The bullets describe a session context and a class reference, not two classes. `both involving classes the JDK compiler cannot name` describes their common constraint accurately. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeCompiler.scala: ########## @@ -0,0 +1,1279 @@ +/* + * 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.catalyst.expressions.codegen + +import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, IOException, StringWriter} +import java.net.{JarURLConnection, URI, URL} +import java.util.Locale +import java.util.concurrent.{Callable, ExecutionException, ExecutorService} +import javax.tools.{Diagnostic, DiagnosticCollector, FileObject, ForwardingJavaFileManager, JavaCompiler, JavaFileManager, JavaFileObject, SimpleJavaFileObject, StandardJavaFileManager, StandardLocation, ToolProvider} + +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +import com.google.common.cache.{Cache, CacheBuilder} +import com.google.common.util.concurrent.Uninterruptibles +import org.codehaus.commons.compiler.{CompileException, InternalCompilerException} +import org.codehaus.janino.ClassBodyEvaluator +import org.codehaus.janino.util.ClassFile +import org.codehaus.janino.util.ClassFile.CodeAttribute + +import org.apache.spark.{JobArtifactSet, SparkEnv, TaskContext, TaskKilledException} +import org.apache.spark.executor.{ExecutorClassLoader, InputMetrics} +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.metrics.source.CodegenMetrics +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Expression, UnsafeArrayData, UnsafeMapData, UnsafeRow} +import org.apache.spark.sql.catalyst.util.{ArrayData, CollationAwareUTF8String, CollationFactory, CollationSupport, MapData} +import org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.Decimal +import org.apache.spark.unsafe.Platform +import org.apache.spark.unsafe.types.{BinaryView, CalendarInterval, TimestampNanosVal, UTF8String, VariantVal} +import org.apache.spark.util.{ParentClassLoader, ThreadUtils, Utils} + +/** + * Backend used to compile generator-produced Java source into a [[GeneratedClass]]. + * + * Two implementations are provided: + * - [[JaninoCodeCompiler]]: the default, uses Janino's `ClassBodyEvaluator`. Very fast. + * - [[JdkCodeCompiler]]: uses `javax.tools.JavaCompiler` from the JDK. Slower (~5x for + * large generated units, 30-300x for small ones), but maintained on the JDK + * release cadence and not subject to Janino's unmaintained-upstream risk. + * + * The backend is selected at compile time via [[SQLConf.CODEGEN_COMPILER]]. + */ +trait CodeCompiler { + /** Backend name as used in `spark.sql.codegen.compiler`. */ + def name: String + + /** + * Compile a generator-produced class body into an instance of the + * [[GeneratedClass]] subclass it defines. + * + * @return the instantiated generated class along with bytecode statistics. + */ + def compile(code: CodeAndComment): (GeneratedClass, ByteCodeStats) +} + +object CodeCompiler extends Logging { + + // Emit log messages under CodeGenerator's logger name: operators and tests + // (SPARK-25113 / SPARK-51527) subscribe to that exact logger for codegen + // compilation events, and the backends are implementation details of + // `CodeGenerator.compile`, so their logs belong under its name. + override protected def logName: String = classOf[CodeGenerator[_, _]].getName + + val JANINO: String = "janino" + val JDK: String = "jdk" + + /** + * Fully-qualified imports made available to generated code by both backends. + * + * For Janino these are passed to `ClassBodyEvaluator.setDefaultImports`. + * For the JDK backend they are rendered into `import` statements inside the + * synthesized compilation unit. + * + * This is the single shared list - anything added here automatically applies to + * both backends. It intentionally excludes + * `org.apache.spark.sql.catalyst.expressions.codegen.GeneratedClass` to avoid + * a name collision with the generated subclass `GeneratedClass`; the extends + * clause uses the fully-qualified name instead. + * + * When adding an entry, keep its SIMPLE name distinct from any `import` line a + * generator emits at the top of a class body (currently only GenerateColumnAccessor + * does this): javac rejects two single-type imports sharing a simple name + * (JLS 7.5.1) while Janino resolves them leniently, so a collision would fail only + * under the JDK backend. + */ + val DefaultImports: Seq[String] = Seq( + classOf[Platform].getName, + classOf[InternalRow].getName, + classOf[UnsafeRow].getName, + classOf[BinaryView].getName, + classOf[UTF8String].getName, + classOf[Decimal].getName, + classOf[CalendarInterval].getName, + classOf[TimestampNanosVal].getName, + classOf[VariantVal].getName, + classOf[ArrayData].getName, + classOf[UnsafeArrayData].getName, + classOf[MapData].getName, + classOf[UnsafeMapData].getName, + classOf[Expression].getName, + classOf[TaskContext].getName, + classOf[TaskKilledException].getName, + classOf[InputMetrics].getName, + classOf[CollationAwareUTF8String].getName, + classOf[CollationFactory].getName, + classOf[CollationSupport].getName, + QueryExecutionErrors.getClass.getName.stripSuffix("$") + ) + + /** + * FQN of the generated class. Must NOT be under the `codegen` package or Janino + * fails with `java.lang.InstantiationException`. The same name is used for both + * backends so generated source, logs, and diagnostics name the same class + * whichever backend compiles it. (Compiled results are NOT shared across + * backends: the compile cache key includes the backend.) + */ + val GeneratedClassName: String = + "org.apache.spark.sql.catalyst.expressions.GeneratedClass" + + def active(): CodeCompiler = active(null) + + /** + * Resolve the active backend for the given generated unit. + * + * The configured backend ([[SQLConf.CODEGEN_COMPILER]]) governs ordinary codegen. The + * exception is codegen the JDK compiler is fundamentally *incapable* of compiling - not + * merely slower at - which is always routed to Janino regardless of the configured + * backend. This is deterministic routing decided up front from the execution context and + * the generated source; it is never a fallback after a failed compile. Two such cases, + * both classes the JDK compiler cannot name that Janino's lenient loader/lexer accepts: + * + * - REPL / interactive sessions (spark-shell `$line*` wrappers, Spark Connect / + * Ammonite session artifacts): reachable only through a runtime class loader and + * carrying self-inconsistent reflection metadata the JDK compiler cannot resolve. + * This arm is context-wide by design: ALL codegen in such a session routes to + * Janino, whether or not the unit references a REPL class, because the reference + * cannot be told from the source text up front. See [[isReplContext]]. + * - A reference to a class nested in a Scala `package object` (binary name + * `a.b.package$Inner`): `package` is a Java reserved word that cannot be spelled as + * an identifier in any form - Java has no backtick/escape, unlike Scala - so javac + * can neither parse `a.b.package.Inner` nor resolve the flat `a.b.package$Inner`. + * See [[requiresJaninoSource]]. + */ + def active(code: CodeAndComment): CodeCompiler = { + val requested = SQLConf.get.codegenCompiler + if (requested != JANINO && isReplContext) { + logReplRoutingOnce() + JaninoCodeCompiler + } else if (requested != JANINO && requiresJaninoSource(code)) { + logPackageObjectRoutingOnce() + JaninoCodeCompiler + } else { + forBackend(requested) + } + } + + // One-time visibility for the deterministic routing above: an operator who set + // `jdk` should be able to tell from the logs why Janino still shows up. + private val replRoutingLogged = new java.util.concurrent.atomic.AtomicBoolean(false) + private def logReplRoutingOnce(): Unit = { + if (replRoutingLogged.compareAndSet(false, true)) { + logInfo(log"REPL / interactive session context detected; codegen is routed to " + + log"Janino although ${MDC(LogKeys.CONFIG, SQLConf.CODEGEN_COMPILER.key)} " + + log"requests another backend (the JDK compiler cannot resolve REPL-defined " + + log"classes). This notice is logged once per JVM.") + } + } + private val packageObjectRoutingLogged = new java.util.concurrent.atomic.AtomicBoolean(false) + private def logPackageObjectRoutingOnce(): Unit = { + if (packageObjectRoutingLogged.compareAndSet(false, true)) { + logInfo(log"Generated code references a Scala package-object class; that unit is " + + log"routed to Janino although ${MDC(LogKeys.CONFIG, SQLConf.CODEGEN_COMPILER.key)} " + + log"requests another backend (`package` is a Java reserved word the JDK compiler " + + log"cannot name). This notice is logged once per JVM.") + } + } + + // A `package` segment in a qualified/binary class name - a Scala `package object`'s nested + // class such as `a.b.package$Inner`. `package` is a Java reserved word the JDK compiler can + // name in no form (parse error as `package.Inner`; unresolvable as the flat `package$Inner`), + // whereas Janino's lexer scans `package$Inner` as one identifier. + // + // `package` is the only keyword scanned for, by design. It is the only Java keyword the + // Scala compiler ever produces in a generated name (from `package object`); a class named + // after any other keyword (`class int`) requires pathological user code. It is also the only + // keyword that is *safe* to scan for: the rest (`int`, `new`, `this`, `return`, `switch`, ...) + // occur as legitimate tokens throughout the generated Java, so matching them would route + // almost all codegen to Janino, whereas a `package` token never appears in a generated class + // body except as such a class reference. (A fully general check would inspect the resolved + // class names rather than the source text, but that information is only available during + // compilation, i.e. after the backend is already chosen.) The lookbehind keeps a legal + // identifier like `mypackage$Inner` from matching; a false positive (e.g. text inside a string + // literal) is harmless - it only picks Janino, a superset of what javac accepts. + private val UnnameablePackageObjectClass = """(?<![\w$])package[.$]""".r + private def requiresJaninoSource(code: CodeAndComment): Boolean = { + // This runs on every compile() call (the result is part of the cache key), so gate + // the regex scan behind an intrinsified substring search: generated bodies almost + // never contain the literal `package` at all, and the regex runs only when they do. + code != null && code.body.contains("package") && + UnnameablePackageObjectClass.findFirstIn(code.body).isDefined + } + + /** + * True when codegen is running in a REPL / interactive context, detected via the three + * mechanisms Spark uses to ship such classes: + * - the active job/session carries a REPL or artifact class-dir URI + * ([[JobArtifactSet.getCurrentJobArtifactState]]'s `replClassDirUri`). This is the + * canonical signal: Spark Connect sets it per session and spark-shell falls back to + * it from `spark.repl.class.uri`. It is a thread-local set around both driver-side + * and executor-side work, so it catches driver-side codegen where no + * `ExecutorClassLoader` is in the loader chain (e.g. a Connect UDF over a local + * relation referencing an Ammonite `$sess` class); or + * - `spark.repl.class.uri` set in the active conf (spark-shell sets this globally); or + * - an [[org.apache.spark.executor.ExecutorClassLoader]] somewhere in the active + * class loader chain (created on executors when a session has such a class URI). + * + * The default (non-REPL) job state has no `replClassDirUri`, so ordinary codegen is not + * affected. Any lookup failure conservatively reports `false`, which preserves the + * configured backend. + */ + private def isReplContext: Boolean = { + def hasArtifactReplUri = + try JobArtifactSet.getCurrentJobArtifactState.exists(_.replClassDirUri.isDefined) + catch { case NonFatal(_) => false } + def confHasReplUri = + try Option(SparkEnv.get).exists(_.conf.contains("spark.repl.class.uri")) + catch { case NonFatal(_) => false } + def eclInChain = + try { + var loader = Utils.getContextOrSparkClassLoader + var found = false + while (loader != null && !found) { + if (loader.isInstanceOf[ExecutorClassLoader]) found = true + loader = loader.getParent + } + found + } catch { + case NonFatal(_) => false + } + hasArtifactReplUri || confHasReplUri || eclInChain + } + + /** + * Get the backend by name. SQLConf already validates the value via `checkValues` + * at config-set time, so unknown names should not reach here in normal use; + * tests may call this directly. When `jdk` is requested but the JDK compiler is + * not present at runtime (a JRE-only image), this logs a warning once and falls + * back to Janino so the query does not fail. + */ + private[codegen] def forBackend(requested: String): CodeCompiler = { + requested.toLowerCase(Locale.ROOT) match { + case JANINO => JaninoCodeCompiler + case JDK if JdkCodeCompiler.isAvailable => JdkCodeCompiler + case JDK => + logJdkUnavailableOnce() + JaninoCodeCompiler + case other => + throw new IllegalArgumentException( + s"Unknown ${SQLConf.CODEGEN_COMPILER.key} backend: $other " + + s"(supported: ${Seq(JANINO, JDK).mkString(", ")})") + } + } + + private val jdkUnavailableWarned = new java.util.concurrent.atomic.AtomicBoolean(false) + private def logJdkUnavailableOnce(): Unit = { + if (jdkUnavailableWarned.compareAndSet(false, true)) { + logWarning(log"${MDC(LogKeys.CONFIG, SQLConf.CODEGEN_COMPILER.key)}=jdk requested " + + log"but javax.tools.JavaCompiler is not available on this runtime " + + log"(JRE-only image?). Falling back to Janino for this JVM.") + } + } + + /** + * Compute bytecode statistics for a set of compiled classes. Both backends + * produce the same map shape (className -> classfile bytes), so the analysis is + * shared. This is the only piece of code that depends on Janino's + * `ClassFile` parser (from the `janino` artifact); it can be swapped for ASM + * later without touching either backend. + */ + private[codegen] def computeByteCodeStats( + classBytecodes: Iterable[(String, Array[Byte])]): ByteCodeStats = { + val perClass = classBytecodes.map { case (_, classBytes) => + val classCodeSize = classBytes.length + 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) + if (byteCodeSize > CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT) { + logInfo(log"Generated method too long to be JIT compiled: " + + log"${MDC(LogKeys.CLASS_NAME, cf.getThisClassName)}." + + log"${MDC(LogKeys.METHOD_NAME, method.getName)} is " + + log"${MDC(LogKeys.BYTECODE_SIZE, byteCodeSize)} bytes") + } + byteCodeSize + } + } + // Use `maxOption` to handle classes with no methods (e.g., a synthetic + // module-info-style class). The original Janino-only code would have raised + // UnsupportedOperationException there; we degrade gracefully to -1 instead. + (methodCodeSizes.maxOption.getOrElse(-1), constPoolSize) + } catch { + case NonFatal(e) => + logWarning("Error calculating stats of compiled class.", e) + (-1, -1) + } + } + + val (maxMethodSizes, constPoolSize) = perClass.unzip + ByteCodeStats( + maxMethodCodeSize = maxMethodSizes.maxOption.getOrElse(-1), + maxConstPoolSize = constPoolSize.maxOption.getOrElse(-1), + // Minus 2 for `GeneratedClass` and an outer-most generated class. + // Both backends wrap the class body in a single outer declaration, so the + // emitted class count has the same shape (1 outer wrapper + K user-declared + // classes) and the offset yields the same value under either backend. + // `max(0, ...)` keeps an unexpected emit shape from going negative. + numInnerClasses = math.max(0, classBytecodes.size - 2)) + } + + /** + * Log the generated source on a compilation failure. Behaviour matches the + * original [[CodeGenerator]] implementation. `maxLines` (the session's + * `loggingMaxLinesForCodegen`) is captured by the CALLER: the JDK backend invokes + * this from its compile worker thread, where `SQLConf.get` would silently return + * the default conf instead of the calling session's. + */ + private[codegen] def logGeneratedCodeOnFailure(code: CodeAndComment, maxLines: Int): Unit = { + val formatted = s"\n${CodeFormatter.format(code, maxLines)}" + if (Utils.isTesting) { + logError(formatted) + } else { + logInfo(formatted) + } + } +} + +/** + * The default backend using Janino's `ClassBodyEvaluator`. + * + * This lifts the previous body of `CodeGenerator.doCompile`, with the only + * changes being: imports moved to `CodeCompiler.DefaultImports`, stats + * computation moved to `CodeCompiler.computeByteCodeStats` (which degrades to + * `-1` for a class with no methods instead of throwing; see its comment). + * Behaviour is otherwise preserved, including the [[ParentClassLoader]] wrapping + * (workaround for the Janino `findIClass` behaviour described in SPARK-15622 / + * SPARK-11636). + */ +object JaninoCodeCompiler extends CodeCompiler with Logging { + + override val name: String = CodeCompiler.JANINO + + // Route source-code/debug log emissions under CodeGenerator's logger name. + override protected def logName: String = classOf[CodeGenerator[_, _]].getName + + override def compile(code: CodeAndComment): (GeneratedClass, ByteCodeStats) = { + val evaluator = new ClassBodyEvaluator() + + // See SPARK-15622 / SPARK-11636 for why this wrapping is required. + val parentClassLoader = new ParentClassLoader(Utils.getContextOrSparkClassLoader) + evaluator.setParentClassLoader(parentClassLoader) + evaluator.setClassName(CodeCompiler.GeneratedClassName) + evaluator.setDefaultImports(CodeCompiler.DefaultImports: _*) + evaluator.setExtendedClass(classOf[GeneratedClass]) + + logBasedOnLevel(SQLConf.get.codegenLogLevel) { + // Only add extra debugging info to byte code when we are going to print the source code. + evaluator.setDebuggingInformation(true, true, false) + log"\n${MDC(LogKeys.CODE, CodeFormatter.format(code))}" + } + + val codeStats = + try { + evaluator.cook("generated.java", code.body) + CodeCompiler.computeByteCodeStats(evaluator.getBytecodes.asScala) + } catch { + case e: InternalCompilerException => + logError("Failed to compile the generated Java code.", e) + CodeCompiler.logGeneratedCodeOnFailure(code, SQLConf.get.loggingMaxLinesForCodegen) + throw QueryExecutionErrors.internalCompilerError(e) + case e: CompileException => + logError("Failed to compile the generated Java code.", e) + CodeCompiler.logGeneratedCodeOnFailure(code, SQLConf.get.loggingMaxLinesForCodegen) + throw QueryExecutionErrors.compilerError(e) + } + + (evaluator.getClazz().getConstructor().newInstance().asInstanceOf[GeneratedClass], codeStats) + } +} + +/** + * Alternative backend using the JDK's `javax.tools.JavaCompiler`. + * + * Wraps the generator-produced class body in a synthesized compilation unit + * (package + imports + `public class GeneratedClass extends ...`) before + * handing it to the compiler. Compiled classes are captured in memory and + * loaded through a private [[ClassLoader]] that mirrors the Janino backend's + * [[ParentClassLoader]] wrapping (SPARK-15622 / SPARK-11636) so behaviour on + * containerised deployments stays consistent. + * + * Class resolution: referenced classes are resolved through the task's context + * [[ClassLoader]] (see [[ClassLoaderFileManager]]) rather than a file-based + * `-classpath`, mirroring how Janino resolves them. This lets the JDK compiler see + * classes that exist only on a runtime loader - REPL-generated, Spark Connect + * session artifacts - and avoids handing javac a giant classpath to index. + * + * Threading: the actual javac invocation runs on a dedicated single-threaded + * executor (see `compileExecutor`). This is required for correctness on Spark + * task threads (jar reads through interruptible NIO channels vs. task + * interruption), and it also confines the shared, not-thread-safe + * [[StandardJavaFileManager]] (used for platform-class lookups) to one + * thread, so no extra locking is needed. Caller threads only build the source and + * capture the context classloader, then await the result. + * + * Resource lifecycle: the shared [[StandardJavaFileManager]] is intentionally + * never closed. Like the compile executor and the per-jar package index, it is + * JVM-lifetime state rather than a per-compile resource, so this is not a leak. + * + * Performance is roughly 5x slower than Janino for large generated units and + * 30-300x slower for small ones. + * The benefit is decoupling Spark from Janino's release cadence. + */ +object JdkCodeCompiler extends CodeCompiler with Logging { + + override val name: String = CodeCompiler.JDK + + // Route source-code/debug log emissions under CodeGenerator's logger name. + override protected def logName: String = classOf[CodeGenerator[_, _]].getName + + /** True if `javax.tools.JavaCompiler` is available on this runtime. */ + lazy val isAvailable: Boolean = ToolProvider.getSystemJavaCompiler != null + + private lazy val compiler: JavaCompiler = { + val c = ToolProvider.getSystemJavaCompiler + require(c != null, + "javax.tools.JavaCompiler is not available; check isAvailable before use") + c + } + + /** + * Shared file manager. `StandardJavaFileManager` is not thread-safe; reuse is safe + * here because every javac invocation runs on the single-threaded [[compileExecutor]]. + */ + private lazy val sharedFileManager: StandardJavaFileManager = + compiler.getStandardFileManager(null, null, null) + + /** + * Compiler options applied to every compilation. + * + * There is deliberately no `--release`/`-source`/`-target`: the compiled class is + * loaded only into the same JVM that produced it (the cache is in-memory; generated + * code travels between JVMs as SOURCE), so the emitted class-file version is always + * consistent with the running runtime and there is no cross-compilation target to + * pin. Pinning `--release` would only reroute the delegate file manager's + * platform-class lookups through `ct.sym`, adding overhead without benefit. + * + * Note there is no `-classpath`: the [[ClassLoaderFileManager]] resolves the + * `CLASS_PATH` location through the compile's parent [[ClassLoader]] (the task's + * context classloader) rather than a file-based classpath. This mirrors how + * Janino resolves referenced classes, so the JDK backend sees exactly what + * Janino would - including classes that exist only on a runtime classloader + * (REPL-generated, Spark Connect session artifacts) and never on + * `java.class.path`. It also avoids handing javac a giant `-classpath` to index, + * which both bloats compiler memory and is brittle to harvest correctly across + * driver / executor / Connect deployments. + */ + private val compileOptions: java.util.List[String] = Seq( + "-proc:none", // skip annotation processing + "-g:none", // skip debug info + "-nowarn", // suppress warnings + "-implicit:none", // do not compile referenced source files + "-Xlint:none" // disable lints + ).asJava + + /** Source-position package name and simple class name derived once. */ + private val packageName: String = + CodeCompiler.GeneratedClassName.substring(0, CodeCompiler.GeneratedClassName.lastIndexOf('.')) + private val simpleName: String = + CodeCompiler.GeneratedClassName.substring(CodeCompiler.GeneratedClassName.lastIndexOf('.') + 1) + + /** Rendered import block, computed once. */ + private val importBlock: String = + CodeCompiler.DefaultImports.map(i => s"import $i;").mkString("\n") + + /** FQN of the abstract base used in the extends clause (avoids import collision). */ + private val extendsFqn: String = classOf[GeneratedClass].getName + + /** + * Wrap a generator-produced class body in a full compilation unit. Class name + * matches Janino's output so logs and diagnostics name the same class whichever + * backend compiled it. `classLoader` resolves the candidate inner-class + * references for the `$`-rewrite (see [[rewriteInnerClassRefs]]); it must be the + * same loader the compile resolves classes through. + */ + private[codegen] def wrapAsCompilationUnit(body: String, classLoader: ClassLoader): String = { + val (extraImports, cleanedBody) = extractLeadingImports(body) + val javacBody = stripFunction1ApplyBridges(cleanedBody) + // Built by plain concatenation, NOT a stripMargin template: stripMargin + // post-processes the final interpolated string, so a generated-body line whose + // first non-blank character is `|` (e.g. a line-wrapped `||` condition) would + // lose that character. No current generator emits such a line, but the unit + // must not depend on that. + s"package $packageName;\n" + + s"$importBlock\n" + + s"$extraImports\n" + + s"public class $simpleName extends $extendsFqn {\n" + + s"${rewriteInnerClassRefs(javacBody, classLoader)}\n" + + "}\n" + } + + // The explicit `scala.Function1` `apply(Object)` bridge that projection codegen emits + // for the Janino backend (see `CodeGenerator.function1ApplyBridge`). javac synthesizes + // this bridge itself for the typed `apply(InternalRow)` override and rejects an explicit + // duplicate with a "name clash" error, so it must be removed before compiling with the + // JDK backend. This pattern matches exactly the shape `function1ApplyBridge` emits + // (whitespace tolerant, `\1` ties the cast operand to the parameter); keep them in sync. + private val Function1ApplyBridgePattern = + ("""(?s)public\s+java\.lang\.Object\s+apply\(\s*java\.lang\.Object\s+(\w+)\s*\)\s*""" + + """\{\s*return\s+apply\(\(\s*InternalRow\s*\)\s*\1\s*\)\s*;\s*\}""").r + + /** Remove the Janino-only Function1 `apply(Object)` bridges so javac does not clash. */ + private[codegen] def stripFunction1ApplyBridges(body: String): String = + if (body.contains("apply(java.lang.Object")) { + Function1ApplyBridgePattern.replaceAllIn(body, "") + } else { + body + } + + /** + * Some generators (e.g. GenerateColumnAccessor) emit `import` statements at + * the top of the class body. Janino's ClassBodyEvaluator treats those as + * imports for the synthesized class, but the JDK compiler rejects imports + * inside a class declaration ("illegal start of type"). Extract any leading + * `import` lines from the body so they can be hoisted into the compilation + * unit header. + */ + private[codegen] def extractLeadingImports(body: String): (String, String) = { + // Fast path: no leading `import` line (every generator but GenerateColumnAccessor). + // Skips the full line-split allocation. + var p = 0 + while (p < body.length && Character.isWhitespace(body.charAt(p))) p += 1 + if (!body.startsWith("import ", p)) return ("", body) + // The `-1` limit keeps trailing empty lines so the reconstruction is faithful. + val lines = body.split("\n", -1) + val imports = new StringBuilder + var i = 0 + var scanning = true + while (scanning && i < lines.length) { + val trimmed = lines(i).trim + if (trimmed.startsWith("import ")) { + imports.append(lines(i)).append('\n') + i += 1 + } else if (trimmed.isEmpty) { + i += 1 + } else { + scanning = false + } + } + if (i == 0) { + ("", body) + } else { + (imports.toString, lines.drop(i).mkString("\n")) + } + } + + /** + * Rewrite JVM-binary inner-class references into the Java-source form the JDK + * compiler accepts. Spark generators emit class names via `Class#getName` in + * many places; for nested classes that yields the binary form (`Outer$Inner`), + * which Janino accepts as a source-level identifier but the JDK compiler does + * not. + * + * The correct source form depends on HOW the class is nested, and that cannot + * be told from the text alone: + * - a regular nested class `Outer$Inner` must be written `Outer.Inner`; + * - a class nested inside a Scala `object` has a binary name whose `$` + * separators include the module suffix (e.g. `Model$SaveLoad$Leaf` where + * `SaveLoad` is an object), and the JDK compiler resolves it ONLY via the + * raw binary name - the dotted canonical form `Model.SaveLoad$.Leaf` makes + * javac reconstruct a non-existent `Model$SaveLoad$$Leaf`. + * Textually `Model$SaveLoad$Leaf` (object-nested) is indistinguishable from + * `A$B$C` (three regular classes) yet they need opposite treatment, so the + * decision is made by resolving each candidate against the compile classpath + * and consulting reflection: `getCanonicalName` is the right source form when + * it is free of `$`, otherwise the binary name is. + * + * For each maximal qualified-name token that contains `$`, we find the longest + * dot-delimited prefix that loads as a class and replace it with that + * reflection-derived name, leaving any trailing member access untouched + * (so `Foo$.MODULE$.apply` and `List$.MODULE$.newBuilder()` resolve correctly). + * Tokens whose prefixes do not resolve - notably references to the + * not-yet-compiled inner classes of the generated unit itself - fall back to a + * conservative regex that dots `$`-before-uppercase, matching the historical + * behaviour for those. + * + * 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. + */ + 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 + } + } + flushCode() + out.toString + } + + /** + * Rewrite the qualified-name tokens of a code span (no literals or comments). + * Runs of `[A-Za-z0-9_$.]` are treated as candidate qualified names; only + * those containing `$` are resolved (others cannot be binary inner-class + * references), and everything else is copied verbatim so whitespace and + * punctuation are preserved exactly. + */ + private def rewriteCodeSpan( + span: String, + classLoader: ClassLoader, + memo: mutable.Map[String, String]): String = { + val sb = new java.lang.StringBuilder(span.length + 16) + val n = span.length + var i = 0 + while (i < n) { + val c = span.charAt(i) + if (isNameStart(c)) { + val start = i + i += 1 + while (i < n && isNamePart(span.charAt(i))) i += 1 + val token = span.substring(start, i) + if (token.indexOf('$') < 0) { + sb.append(token) + } else { + sb.append(memo.getOrElseUpdate(token, rewriteQualifiedName(token, classLoader))) + } + } else { + sb.append(c) + i += 1 + } + } + sb.toString + } + + private def isNameStart(c: Char): Boolean = + (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_' || c == '$' + + private def isNamePart(c: Char): Boolean = + isNameStart(c) || (c >= '0' && c <= '9') || c == '.' + + /** + * Resolve a `$`-containing qualified token to a JDK-compiler-acceptable form by + * finding the longest dot-delimited prefix that loads as a class and replacing + * it with its reflection-derived source name, keeping any trailing member + * access. Falls back to the conservative `$`-before-uppercase regex when no + * prefix resolves (e.g. inner classes of the not-yet-compiled generated unit). + */ + private def rewriteQualifiedName(token: String, classLoader: ClassLoader): String = { + val parts = token.split('.') + var k = parts.length + while (k >= 1) { + val prefix = parts.iterator.take(k).mkString(".") + // Only a prefix that itself contains `$` can be a binary inner-class name. + if (prefix.indexOf('$') >= 0) { + resolveSourceName(prefix, classLoader) match { + case Some(sourceName) => + val rest = parts.iterator.drop(k).mkString(".") + val resolved = if (rest.isEmpty) sourceName else s"$sourceName.$rest" + // `split('.')` drops a trailing empty segment, so a token that ends in `.` + // (member access wrapped onto the next line) must get its dot restored. + return if (token.endsWith(".")) resolved + "." else resolved + case None => // try a shorter prefix + } + } + k -= 1 + } + InnerClassRefPattern.replaceAllIn(token, ".") + } + + /** + * Load `binaryName` without initializing it and return the source name the JDK + * compiler accepts: the canonical name when it is a plain dotted identifier + * name (regular nesting, e.g. `java.util.Map.Entry`), otherwise the binary + * name itself. The binary name is required for classes nested in Scala objects + * and for Scala companion-object classes, whose canonical form carries a + * module `$` that javac cannot resolve, and for Scala operator-named classes + * (e.g. `scala.collection.immutable.::`) whose canonical form is not a valid + * Java identifier - in all those cases the binary name is itself a legal Java + * type reference. Returns None when the name is not a loadable class. + * + * The canonical name is also rejected when it is not a faithful rename of the + * binary name - it must keep the same package. Scala REPL classes (e.g. + * `$line21.$read$$iw$TestCaseClass`) report a misleading `getCanonicalName` that + * drops the package and returns just the simple name (`TestCaseClass`); using it + * would corrupt the reference into an unqualified one javac cannot resolve. + */ + private def resolveSourceName(binaryName: String, classLoader: ClassLoader): Option[String] = { + try { + // scalastyle:off classforname + // Load with the exact loader passed in (the task's context loader), not the Spark + // class loader, so the JDK compiler sees what the runtime would; Utils.classForName + // cannot target an arbitrary loader. + val loaded = Class.forName(binaryName, false, classLoader) + // scalastyle:on classforname + val cls = nameableSupertype(loaded) Review Comment: This narrowing is only safe when every generated member access is declared by the chosen supertype, but nothing here enforces that. An anonymous `ObjectType` that adds a method will be rewritten to its superclass or first interface, so javac can no longer resolve that call. Please preserve a declared nameable reference type during generation, or route the unit to Janino when the concrete type is anonymous/local and its accessed members cannot be proven available on the replacement type. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
