Add enabled config

Author: andygrove

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -654,6 +654,30 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(defaultValue)
   }
+
+  def isExprEnabled(name: String, conf: SQLConf = SQLConf.get): Boolean = {
+    val value = getExprEnabledConfigKey(name)
+    // all expressions are enabled by default
+    val key = value
+    if (conf.contains(key)) {
+      conf.getConfString(key) == "true"
+    } else {
+      true
+    }
+  }
+
+  def getExprEnabledConfigKey(name: String): String = {
+    s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.$name.enabled"
+  }
+
+  def isExprAllowIncompat(name: String, conf: SQLConf = SQLConf.get): Boolean = {
+    val key = getExprAllowIncompatConfigKey(name)
+    conf.contains(key) && conf.getConfString(key) == "true"
+  }
+
+  def getExprAllowIncompatConfigKey(name: String): String = {
+    s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.$name.allowIncompatible"
+  }
 }
 
 object ConfigHelpers {

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -598,17 +598,21 @@ object QueryPlanSerde extends Logging with CometExprShim {
     val conf = SQLConf.get
 
     def convert[T <: Expression](expr: T, handler: CometExpressionSerde[T]): Option[Expr] = {
+      val exprConfName = handler.getExprConfigName(expr)
+      if (!CometConf.isExprEnabled(exprConfName)) {
+        withInfo(
+          expr,
+          "Expression support is disabled. Set " +
+            s"${CometConf.getExprEnabledConfigKey(exprConfName)}=true to enable it.")
+        return None
+      }
       handler.getSupportLevel(expr) match {
         case Unsupported(notes) =>
           withInfo(expr, notes.getOrElse(""))
           None
         case Incompatible(notes) =>
-          val exprConfName = handler.getExprConfigName(expr)
-          val exprIncompatConf = s"${CometConf.COMET_EXPR_CONFIG_PREFIX}.$exprConfName" +
-            ".allowIncompatible"
-          val exprEnabled = conf.contains(exprIncompatConf) &&
-            conf.getConfString(exprIncompatConf) == "true"
-          if (exprEnabled || CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.get()) {
+          val exprAllowIncompat = CometConf.isExprAllowIncompat(exprConfName)
+          if (exprAllowIncompat || CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.get()) {
             if (notes.isDefined) {
               logWarning(
                 s"Comet supports $expr when ${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true " +
@@ -620,7 +624,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
             withInfo(
               expr,
               s"$expr is not fully compatible with Spark$optionalNotes. To enable it anyway, " +
-                s"set $exprIncompatConf=true, or set " +
+                s"set ${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true, or set " +
                 s"${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true to enable all " +
                 s"incompatible expressions. ${CometConf.COMPAT_GUIDE}.")
             None