[SPARK-23711][SPARK-25140][SQL] Catch correct exceptions when expr codegen fails

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

@@ -17,24 +17,10 @@
 
 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.internal.Logging
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.util.Utils
 
-/**
- * Catches compile error during code generation.
- */
-object CodegenError {
-  def unapply(throwable: Throwable): Option[Exception] = throwable match {
-    case e: InternalCompilerException => Some(e)
-    case e: CompileException => Some(e)
-    case _ => None
-  }
-}
-
 /**
  * Defines values for `SQLConf` config of fallback mode. Use for test only.
  */
@@ -47,7 +33,7 @@ object CodegenObjectFactoryMode extends Enumeration {
  * error happens, it can fallback to interpreted implementation. In tests, we can use a SQL config
  * `SQLConf.CODEGEN_FACTORY_MODE` to control fallback behavior.
  */
-abstract class CodeGeneratorWithInterpretedFallback[IN, OUT] {
+abstract class CodeGeneratorWithInterpretedFallback[IN, OUT] extends Logging {
 
   def createObject(in: IN): OUT = {
     // We are allowed to choose codegen-only or no-codegen modes if under tests.
@@ -63,7 +49,10 @@ abstract class CodeGeneratorWithInterpretedFallback[IN, OUT] {
         try {
           createCodeGeneratedObject(in)
         } catch {
-          case CodegenError(_) => createInterpretedObject(in)
+          case _: Exception =>
+            // We should have already seen the error message in `CodeGenerator`
+            logWarning("Expr codegen error and falling back to interpreter mode")
+            createInterpretedObject(in)
         }
     }
   }

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

@@ -180,7 +180,10 @@ object UnsafeProjection
     try {
       GenerateUnsafeProjection.generate(unsafeExprs, subexpressionEliminationEnabled)
     } catch {
-      case CodegenError(_) => InterpretedUnsafeProjection.createProjection(unsafeExprs)
+      case _: Exception =>
+        // We should have already seen the error message in `CodeGenerator`
+        logWarning("Expr codegen error and falling back to interpreter mode")
+        InterpretedUnsafeProjection.createProjection(unsafeExprs)
     }
   }
 }

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

@@ -17,17 +17,33 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import java.util.concurrent.ExecutionException
+
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodeAndComment, CodeGenerator}
 import org.apache.spark.sql.catalyst.plans.PlanTestBase
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{IntegerType, LongType}
+import org.apache.spark.sql.types.IntegerType
 
 class CodeGeneratorWithInterpretedFallbackSuite extends SparkFunSuite with PlanTestBase {
 
-  test("UnsafeProjection with codegen factory mode") {
-    val input = Seq(LongType, IntegerType)
-      .zipWithIndex.map(x => BoundReference(x._2, x._1, true))
+  object FailedCodegenProjection
+      extends CodeGeneratorWithInterpretedFallback[Seq[Expression], UnsafeProjection] {
+
+    override protected def createCodeGeneratedObject(in: Seq[Expression]): UnsafeProjection = {
+      val invalidCode = new CodeAndComment("invalid code", Map.empty)
+      // We assume this compilation throws an exception
+      CodeGenerator.compile(invalidCode)
+      null
+    }
+
+    override protected def createInterpretedObject(in: Seq[Expression]): UnsafeProjection = {
+      InterpretedUnsafeProjection.createProjection(in)
+    }
+  }
 
+  test("UnsafeProjection with codegen factory mode") {
+    val input = Seq(BoundReference(0, IntegerType, nullable = true))
     val codegenOnly = CodegenObjectFactoryMode.CODEGEN_ONLY.toString
     withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> codegenOnly) {
       val obj = UnsafeProjection.createObject(input)
@@ -40,4 +56,24 @@ class CodeGeneratorWithInterpretedFallbackSuite extends SparkFunSuite with PlanT
       assert(obj.isInstanceOf[InterpretedUnsafeProjection])
     }
   }
+
+  test("fallback to the interpreter mode") {
+    val input = Seq(BoundReference(0, IntegerType, nullable = true))
+    val fallback = CodegenObjectFactoryMode.FALLBACK.toString
+    withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> fallback) {
+      val obj = FailedCodegenProjection.createObject(input)
+      assert(obj.isInstanceOf[InterpretedUnsafeProjection])
+    }
+  }
+
+  test("codegen failures in the CODEGEN_ONLY mode") {
+    val errMsg = intercept[ExecutionException] {
+      val input = Seq(BoundReference(0, IntegerType, nullable = true))
+      val codegenOnly = CodegenObjectFactoryMode.CODEGEN_ONLY.toString
+      withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> codegenOnly) {
+        FailedCodegenProjection.createObject(input)
+      }
+    }.getMessage
+    assert(errMsg.contains("failed to compile: org.codehaus.commons.compiler.CompileException:"))
+  }
 }