feat: Add config option to log fallback reasons (#2154)

Author: andygrove

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

@@ -482,6 +482,12 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_LOG_FALLBACK_REASONS: ConfigEntry[Boolean] =
+    conf("spark.comet.logFallbackReasons.enabled")
+      .doc("When this setting is enabled, Comet will log warnings for all fallback reasons.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_EXPLAIN_FALLBACK_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.explainFallback.enabled")
       .doc(

docs/source/user-guide/configs.md

@@ -68,6 +68,7 @@ Comet provides the following configuration settings.
 | spark.comet.explain.verbose.enabled | When this setting is enabled, Comet will provide a verbose tree representation of the extended information. | false |
 | spark.comet.explainFallback.enabled | When this setting is enabled, Comet will provide logging explaining the reason(s) why a query stage cannot be executed natively. Set this to false to reduce the amount of logging. | false |
 | spark.comet.expression.allowIncompatible | Comet is not currently fully compatible with Spark for all expressions. Set this config to true to allow them anyway. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html). | false |
+| spark.comet.logFallbackReasons.enabled | When this setting is enabled, Comet will log warnings for all fallback reasons. | false |
 | spark.comet.memory.overhead.factor | Fraction of executor memory to be allocated as additional memory for Comet when running Spark in on-heap mode. For more information, refer to the Comet Tuning Guide (https://datafusion.apache.org/comet/user-guide/tuning.html). | 0.2 |
 | spark.comet.memory.overhead.min | Minimum amount of additional memory to be allocated per executor process for Comet, in MiB, when running Spark in on-heap mode. For more information, refer to the Comet Tuning Guide (https://datafusion.apache.org/comet/user-guide/tuning.html). | 402653184b |
 | spark.comet.memoryOverhead | The amount of additional memory to be allocated per executor process for Comet, in MiB, when running Spark in on-heap mode. This config is optional. If this is not specified, it will be set to `spark.comet.memory.overhead.factor` * `spark.executor.memory`. For more information, refer to the Comet Tuning Guide (https://datafusion.apache.org/comet/user-guide/tuning.html). | |

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -366,6 +366,11 @@ object CometSparkSessionExtensions extends Logging {
    *   The node with information (if any) attached
    */
   def withInfos[T <: TreeNode[_]](node: T, info: Set[String], exprs: T*): T = {
+    if (CometConf.COMET_LOG_FALLBACK_REASONS.get()) {
+      for (reason <- info) {
+        logWarning(s"Comet cannot accelerate ${node.getClass.getSimpleName} because: $reason")
+      }
+    }
     val existingNodeInfos = node.getTagValue(CometExplainInfo.EXTENSION_INFO)
     val newNodeInfo = (existingNodeInfos ++ exprs
       .flatMap(_.getTagValue(CometExplainInfo.EXTENSION_INFO))).flatten.toSet

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -789,7 +789,7 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
     }
 
     if (!isCometPlan(s.child)) {
-      withInfo(s, "Child {s.child.getClass.getName} is not native")
+      withInfo(s, s"Child ${s.child.getClass.getName} is not native")
       return false
     }
 
@@ -841,12 +841,12 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
     }
 
     if (isShuffleOperator(s.child)) {
-      withInfo(s, "Child {s.child.getClass.getName} is a shuffle operator")
+      withInfo(s, s"Child ${s.child.getClass.getName} is a shuffle operator")
       return false
     }
 
     if (!(!s.child.supportsColumnar || isCometPlan(s.child))) {
-      withInfo(s, "Child {s.child.getClass.getName} is a neither row-based or a Comet operator")
+      withInfo(s, s"Child ${s.child.getClass.getName} is a neither row-based or a Comet operator")
       return false
     }
 

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

@@ -192,10 +192,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[Corr] -> CometCorr,
     classOf[BloomFilterAggregate] -> CometBloomFilterAggregate)
 
-  def emitWarning(reason: String): Unit = {
-    logWarning(s"Comet native execution is disabled due to: $reason")
-  }
-
   def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
@@ -207,8 +203,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
       supportedDataType(a.elementType, allowComplex)
     case m: MapType if allowComplex =>
       supportedDataType(m.keyType, allowComplex) && supportedDataType(m.valueType, allowComplex)
-    case dt =>
-      emitWarning(s"unsupported Spark data type: $dt")
+    case _ =>
       false
   }
 
@@ -237,7 +232,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
       case _: MapType => 15
       case _: StructType => 16
       case dt =>
-        emitWarning(s"Cannot serialize Spark data type: $dt")
+        logWarning(s"Cannot serialize Spark data type: $dt")
         return None
     }
 
@@ -503,9 +498,10 @@ object QueryPlanSerde extends Logging with CometExprShim {
       case Some(handler) =>
         handler.convert(aggExpr, fn, inputs, binding, conf)
       case _ =>
-        val msg = s"unsupported Spark aggregate function: ${fn.prettyName}"
-        emitWarning(msg)
-        withInfo(aggExpr, msg, fn.children: _*)
+        withInfo(
+          aggExpr,
+          s"unsupported Spark aggregate function: ${fn.prettyName}",
+          fn.children: _*)
         None
     }
   }
@@ -1365,18 +1361,23 @@ object QueryPlanSerde extends Logging with CometExprShim {
           ExprOuterClass.Expr.newBuilder().setNormalizeNanAndZero(builder).build()
         }
 
-      case s @ execution.ScalarSubquery(_, _) if supportedDataType(s.dataType) =>
-        val dataType = serializeDataType(s.dataType)
-        if (dataType.isEmpty) {
-          withInfo(s, s"Scalar subquery returns unsupported datatype ${s.dataType}")
-          return None
-        }
+      case s @ execution.ScalarSubquery(_, _) =>
+        if (supportedDataType(s.dataType)) {
+          val dataType = serializeDataType(s.dataType)
+          if (dataType.isEmpty) {
+            withInfo(s, s"Scalar subquery returns unsupported datatype ${s.dataType}")
+            return None
+          }
 
-        val builder = ExprOuterClass.Subquery
-          .newBuilder()
-          .setId(s.exprId.id)
-          .setDatatype(dataType.get)
-        Some(ExprOuterClass.Expr.newBuilder().setSubquery(builder).build())
+          val builder = ExprOuterClass.Subquery
+            .newBuilder()
+            .setId(s.exprId.id)
+            .setDatatype(dataType.get)
+          Some(ExprOuterClass.Expr.newBuilder().setSubquery(builder).build())
+        } else {
+          withInfo(s, s"Unsupported data type: ${s.dataType}")
+          None
+        }
 
       case UnscaledValue(child) =>
         val childExpr = exprToProtoInternal(child, inputs, binding)
@@ -1777,10 +1778,9 @@ object QueryPlanSerde extends Logging with CometExprShim {
 
         } else {
           // There are unsupported scan type
-          val msg =
-            s"unsupported Comet operator: ${op.nodeName}, due to unsupported data types above"
-          emitWarning(msg)
-          withInfo(op, msg)
+          withInfo(
+            op,
+            s"unsupported Comet operator: ${op.nodeName}, due to unsupported data types above")
           None
         }
 
@@ -1952,9 +1952,10 @@ object QueryPlanSerde extends Logging with CometExprShim {
           val attributes = groupingExpressions.map(_.toAttribute) ++ aggregateAttributes
           val resultExprs = resultExpressions.map(exprToProto(_, attributes))
           if (resultExprs.exists(_.isEmpty)) {
-            val msg = s"Unsupported result expressions found in: ${resultExpressions}"
-            emitWarning(msg)
-            withInfo(op, msg, resultExpressions: _*)
+            withInfo(
+              op,
+              s"Unsupported result expressions found in: $resultExpressions",
+              resultExpressions: _*)
             return None
           }
           hashAggBuilder.addAllResultExprs(resultExprs.map(_.get).asJava)
@@ -1996,9 +1997,10 @@ object QueryPlanSerde extends Logging with CometExprShim {
               val attributes = groupingExpressions.map(_.toAttribute) ++ aggregateAttributes
               val resultExprs = resultExpressions.map(exprToProto(_, attributes))
               if (resultExprs.exists(_.isEmpty)) {
-                val msg = s"Unsupported result expressions found in: ${resultExpressions}"
-                emitWarning(msg)
-                withInfo(op, msg, resultExpressions: _*)
+                withInfo(
+                  op,
+                  s"Unsupported result expressions found in: $resultExpressions",
+                  resultExpressions: _*)
                 return None
               }
               hashAggBuilder.addAllResultExprs(resultExprs.map(_.get).asJava)
@@ -2172,6 +2174,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
           op.output.forall(a => supportedDataType(a.dataType, allowComplex = true))
 
         if (!supportedTypes) {
+          withInfo(op, "Unsupported data type")
           return None
         }
 
@@ -2197,10 +2200,9 @@ object QueryPlanSerde extends Logging with CometExprShim {
           Some(builder.setScan(scanBuilder).build())
         } else {
           // There are unsupported scan type
-          val msg =
-            s"unsupported Comet operator: ${op.nodeName}, due to unsupported data types above"
-          emitWarning(msg)
-          withInfo(op, msg)
+          withInfo(
+            op,
+            s"unsupported Comet operator: ${op.nodeName}, due to unsupported data types above")
           None
         }
 
@@ -2222,9 +2224,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
             //  1. it is not Spark shuffle operator, which is handled separately
             //  2. it is not a Comet operator
             if (!op.nodeName.contains("Comet") && !op.isInstanceOf[ShuffleExchangeExec]) {
-              val msg = s"unsupported Spark operator: ${op.nodeName}"
-              emitWarning(msg)
-              withInfo(op, msg)
+              withInfo(op, s"unsupported Spark operator: ${op.nodeName}")
             }
             None
         }