Don't access SQLConf on task execution.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CodegenObjectFactory.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions
 import org.codehaus.commons.compiler.CompileException
 import org.codehaus.janino.InternalCompilerException
 
+import org.apache.spark.TaskContext
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.util.Utils
 
@@ -34,23 +35,36 @@ object CodegenError {
   }
 }
 
+/**
+ * Defines values for `SQLConf` config of fallback mode. Use for test only.
+ */
+object CodegenObjectFactoryMode extends Enumeration {
+  val AUTO, CODEGEN_ONLY, NO_CODEGEN = Value
+
+  def currentMode: CodegenObjectFactoryMode.Value = {
+    // If we weren't on task execution, accesses that config.
+    if (TaskContext.get == null) {
+      val config = SQLConf.get.getConf(SQLConf.CODEGEN_OBJECT_FALLBACK)
+      CodegenObjectFactoryMode.withName(config)
+    } else {
+      CodegenObjectFactoryMode.AUTO
+    }
+  }
+}
+
 /**
  * A factory which can be used to create objects that have both codegen and interpreted
  * implementations. This tries to create codegen object first, if any compile error happens,
  * it fallbacks to interpreted version.
  */
 abstract class CodegenObjectFactory[IN, OUT] {
 
-  // Creates wanted object. First trying codegen implementation. If any compile error happens,
-  // fallbacks to interpreted version.
   def createObject(in: IN): OUT = {
-    val fallbackMode = SQLConf.get.getConf(SQLConf.CODEGEN_OBJECT_FALLBACK)
-    // Only in tests, we can use `SQLConf.CODEGEN_OBJECT_FALLBACK` to choose codegen/interpreted
-    // only path.
-    if (Utils.isTesting && fallbackMode != "fallback") {
-      fallbackMode match {
-        case "codegen-only" => createCodeGeneratedObject(in)
-        case "interpreted-only" => createInterpretedObject(in)
+    // We are allowed to choose codegen-only or no-codegen modes if under tests.
+    if (Utils.isTesting && CodegenObjectFactoryMode.currentMode != CodegenObjectFactoryMode.AUTO) {
+      CodegenObjectFactoryMode.currentMode match {
+        case CodegenObjectFactoryMode.CODEGEN_ONLY => createCodeGeneratedObject(in)
+        case CodegenObjectFactoryMode.NO_CODEGEN => createInterpretedObject(in)
       }
     } else {
       try {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -32,6 +32,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
 import org.apache.spark.network.util.ByteUnit
 import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
 import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator
 import org.apache.spark.util.Utils
 
@@ -688,14 +689,14 @@ object SQLConf {
 
   val CODEGEN_OBJECT_FALLBACK = buildConf("spark.sql.test.codegenObject.fallback")
     .doc("Determines the behavior of any factories extending `CodegenObjectFactory`" +
-      " during tests. `fallback` means trying codegen first and then fallbacking to" +
-      "interpreted if any compile error happens. Disabling fallback if `codegen-only`." +
-      "`interpreted-only` skips codegen and goes interpreted path always. Note that" +
-      "this config works only for tests. In production it always runs with `fallback` mode")
+      " during tests. `AUTO` means trying codegen first and then fallbacking to" +
+      "interpreted if any compile error happens. Disabling fallback if `CODEGEN_ONLY`." +
+      "`NO_CODEGEN` skips codegen and goes interpreted path always. Note that" +
+      "this config works only for tests.")
     .internal()
     .stringConf
-    .checkValues(Set("fallback", "codegen-only", "interpreted-only"))
-    .createWithDefault("fallback")
+    .checkValues(CodegenObjectFactoryMode.values.map(_.toString))
+    .createWithDefault(CodegenObjectFactoryMode.AUTO.toString)
 
   val CODEGEN_FALLBACK = buildConf("spark.sql.codegen.fallback")
     .internal()

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CodegenObjectFactorySuite.scala

@@ -29,7 +29,7 @@ class CodegenObjectFactorySuite extends SparkFunSuite with PlanTestBase {
   private def testCodegenFactory[IN, OUT](factory: CodegenObjectFactory[IN, OUT],
       input: IN, checkerForCodegen: OUT => Unit, checkerForInterpreted: OUT => Unit) = {
 
-    val modes = Seq("codegen-only", "interpreted-only")
+    val modes = Seq("CODEGEN_ONLY", "NO_CODEGEN")
       .zip(Seq(checkerForCodegen, checkerForInterpreted))
 
     for ((fallbackMode, checker) <- modes) {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvalHelper.scala

@@ -197,7 +197,7 @@ trait ExpressionEvalHelper extends GeneratorDrivenPropertyChecks with PlanTestBa
       expression: Expression,
       expected: Any,
       inputRow: InternalRow = EmptyRow): Unit = {
-    for (fallbackMode <- Seq("codegen-only", "interpreted-only")) {
+    for (fallbackMode <- Seq("CODEGEN_ONLY", "NO_CODEGEN")) {
       withSQLConf(SQLConf.CODEGEN_OBJECT_FALLBACK.key -> fallbackMode) {
         val factory = UnsafeProjection
         val unsafeRow = evaluateWithUnsafeProjection(expression, inputRow, factory)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala

@@ -40,7 +40,7 @@ class UnsafeRowConverterSuite extends SparkFunSuite with Matchers with PlanTestB
     f: UnsafeProjection.type => Unit): Unit = {
     val factory = UnsafeProjection
     test(name) {
-      for (fallbackMode <- Seq("codegen-only", "interpreted-only")) {
+      for (fallbackMode <- Seq("CODEGEN_ONLY", "NO_CODEGEN")) {
         withSQLConf(SQLConf.CODEGEN_OBJECT_FALLBACK.key -> fallbackMode) {
           f(factory)
         }