feat: Improve some confusing fallback reasons

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

@@ -396,6 +396,13 @@ object CometSparkSessionExtensions extends Logging {
     withInfos(node, Set.empty, exprs: _*)
   }
 
+  /**
+   * Checks whether a TreeNode has any explain information attached
+   */
+  def hasExplainInfo(node: TreeNode[_]): Boolean = {
+    node.getTagValue(CometExplainInfo.EXTENSION_INFO).exists(_.nonEmpty)
+  }
+
   // Helper to reduce boilerplate
   def createMessage(condition: Boolean, message: => String): Option[String] = {
     if (condition) {

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

@@ -540,8 +540,13 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
             // these cases specially here so we do not add a misleading 'info' message
             op
           case _ =>
-            // An operator that is not supported by Comet
-            withInfo(op, s"${op.nodeName} is not supported")
+            if (!hasExplainInfo(op)) {
+              // An operator that is not supported by Comet
+              withInfo(op, s"${op.nodeName} is not supported")
+            } else {
+              // Already has fallback reason, do not override it
+              op
+            }
         }
     }
   }
@@ -741,7 +746,7 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
         s"Comet shuffle is not enabled: ${COMET_EXEC_SHUFFLE_ENABLED.key} is not enabled")
       false
     } else if (!isCometShuffleManagerEnabled(op.conf)) {
-      withInfo(op, s"spark.shuffle.manager is not set to ${CometShuffleManager.getClass.getName}")
+      withInfo(op, s"spark.shuffle.manager is not set to ${classOf[CometShuffleManager].getName}")
       false
     } else {
       true

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

@@ -64,59 +64,64 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
   }
 
   private def _apply(plan: SparkPlan): SparkPlan = {
-    if (!isCometLoaded(conf) || !isCometScanEnabled(conf)) {
-      if (!isCometLoaded(conf)) {
-        withInfo(plan, "Comet is not enabled")
-      } else if (!isCometScanEnabled(conf)) {
-        withInfo(plan, "Comet Scan is not enabled")
-      }
-      plan
-    } else {
+    if (!isCometLoaded(conf)) return plan
 
-      def hasMetadataCol(plan: SparkPlan): Boolean = {
-        plan.expressions.exists(_.exists {
-          case a: Attribute =>
-            a.isMetadataCol
-          case _ => false
-        })
-      }
+    def isSupportedScanNode(plan: SparkPlan): Boolean = plan match {
+      case _: FileSourceScanExec => true
+      case _: BatchScanExec => true
+      case _ => false
+    }
 
-      def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = {
-        // List of Iceberg metadata tables:
-        // https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables
-        val metadataTableSuffix = Set(
-          "history",
-          "metadata_log_entries",
-          "snapshots",
-          "entries",
-          "files",
-          "manifests",
-          "partitions",
-          "position_deletes",
-          "all_data_files",
-          "all_delete_files",
-          "all_entries",
-          "all_manifests")
-
-        metadataTableSuffix.exists(suffix => scanExec.table.name().endsWith(suffix))
-      }
+    def hasMetadataCol(plan: SparkPlan): Boolean = {
+      plan.expressions.exists(_.exists {
+        case a: Attribute =>
+          a.isMetadataCol
+        case _ => false
+      })
+    }
 
-      plan.transform {
-        case scan if hasMetadataCol(scan) =>
-          withInfo(scan, "Metadata column is not supported")
+    def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = {
+      // List of Iceberg metadata tables:
+      // https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables
+      val metadataTableSuffix = Set(
+        "history",
+        "metadata_log_entries",
+        "snapshots",
+        "entries",
+        "files",
+        "manifests",
+        "partitions",
+        "position_deletes",
+        "all_data_files",
+        "all_delete_files",
+        "all_entries",
+        "all_manifests")
+
+      metadataTableSuffix.exists(suffix => scanExec.table.name().endsWith(suffix))
+    }
 
-        // data source V1
-        case scanExec: FileSourceScanExec =>
-          transformV1Scan(scanExec)
+    def transformScan(plan: SparkPlan): SparkPlan = plan match {
+      case scan if !isCometScanEnabled(conf) =>
+        withInfo(scan, "Comet Scan is not enabled")
 
-        // data source V2
-        case scanExec: BatchScanExec =>
-          if (isIcebergMetadataTable(scanExec)) {
-            withInfo(scanExec, "Iceberg Metadata tables are not supported")
-          } else {
-            transformV2Scan(scanExec)
-          }
-      }
+      case scan if hasMetadataCol(scan) =>
+        withInfo(scan, "Metadata column is not supported")
+
+      // data source V1
+      case scanExec: FileSourceScanExec =>
+        transformV1Scan(scanExec)
+
+      // data source V2
+      case scanExec: BatchScanExec =>
+        if (isIcebergMetadataTable(scanExec)) {
+          withInfo(scanExec, "Iceberg Metadata tables are not supported")
+        } else {
+          transformV2Scan(scanExec)
+        }
+    }
+
+    plan.transform {
+      case scan if isSupportedScanNode(scan) => transformScan(scan)
     }
   }