perf: short-circuit repeat serde checks via per-handler attempt tag

Record the handler class name on operators whose getSupportLevel returned
Unsupported or Incompatible (without allowIncompat), so that later AQE
stage-prep passes skip re-running the same check. The tag is keyed per
handler, so other handlers on the same node (e.g. CometSparkToColumnarExec
on a scan tagged by CometScanRule) are unaffected.

Closes #3990

Author: andygrove

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

@@ -214,6 +214,15 @@ object CometCoverageStats {
 object CometExplainInfo {
   val EXTENSION_INFO = new TreeNodeTag[Set[String]]("CometExtensionInfo")
 
+  /**
+   * Records handler class names whose `getSupportLevel` has already returned `Unsupported` or
+   * `Incompatible` (without `allowIncompat`) on a given operator, so that repeat invocations of
+   * the same handler on the same node during later rule passes can short-circuit without
+   * re-running the check. Orthogonal to [[EXTENSION_INFO]]; keyed per handler so other handlers
+   * on the same node are unaffected.
+   */
+  val FAILED_HANDLERS = new TreeNodeTag[Set[String]]("CometFailedHandlers")
+
   def getActualPlan(node: TreeNode[_]): TreeNode[_] = {
     node match {
       case p: AdaptiveSparkPlanExec => getActualPlan(p.executedPlan)

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

@@ -480,8 +480,13 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
   }
 
   /** Convert a Spark plan to a Comet plan using the specified serde handler */
-  private def convertToComet(op: SparkPlan, handler: CometOperatorSerde[_]): Option[SparkPlan] = {
+  private[rules] def convertToComet(
+      op: SparkPlan,
+      handler: CometOperatorSerde[_]): Option[SparkPlan] = {
     val serde = handler.asInstanceOf[CometOperatorSerde[SparkPlan]]
+    if (hasFailedHandler(op, handler)) {
+      return None
+    }
     if (isOperatorEnabled(serde, op)) {
       // For operators that require native children (like writes), check if all data-producing
       // children are CometNativeExec. This prevents runtime failures when the native operator
@@ -522,6 +527,7 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
       handler.getSupportLevel(op) match {
         case Unsupported(notes) =>
           withInfo(op, notes.getOrElse(""))
+          recordFailedHandler(op, handler)
           false
         case Incompatible(notes) =>
           val allowIncompat = CometConf.isOperatorAllowIncompat(opName)
@@ -540,6 +546,7 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
               s"$opName is not fully compatible with Spark$optionalNotes. " +
                 s"To enable it anyway, set $incompatConf=true. " +
                 s"${CometConf.COMPAT_GUIDE}.")
+            recordFailedHandler(op, handler)
             false
           }
         case Compatible(notes) =>
@@ -557,6 +564,16 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
     }
   }
 
+  private def hasFailedHandler(op: SparkPlan, handler: CometOperatorSerde[_]): Boolean = {
+    op.getTagValue(CometExplainInfo.FAILED_HANDLERS)
+      .exists(_.contains(handler.getClass.getName))
+  }
+
+  private def recordFailedHandler(op: SparkPlan, handler: CometOperatorSerde[_]): Unit = {
+    val existing = op.getTagValue(CometExplainInfo.FAILED_HANDLERS).getOrElse(Set.empty[String])
+    op.setTagValue(CometExplainInfo.FAILED_HANDLERS, existing + handler.getClass.getName)
+  }
+
   private def shouldApplySparkToColumnar(conf: SQLConf, op: SparkPlan): Boolean = {
     // Only consider converting leaf nodes to columnar currently, so that all the following
     // operators can have a chance to be converted to columnar. Leaf operators that output

spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala

@@ -30,7 +30,9 @@ import org.apache.spark.sql.execution.aggregate.HashAggregateExec
 import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleExchangeExec}
 import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
 
-import org.apache.comet.CometConf
+import org.apache.comet.{CometConf, ConfigEntry}
+import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass, SupportLevel, Unsupported}
+import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator}
 
 /**
@@ -206,6 +208,67 @@ class CometExecRuleSuite extends CometTestBase {
     }
   }
 
+  test("convertToComet short-circuits repeat call for a handler that already failed") {
+    withTempView("test_data") {
+      createTestDataFrame.createOrReplaceTempView("test_data")
+
+      val sparkPlan = createSparkPlan(spark, "SELECT id FROM test_data WHERE id > 0")
+      val op = sparkPlan.collectFirst { case f: FilterExec => f }.get
+
+      val stub = new StubUnsupportedHandlerA
+      val rule = CometExecRule(spark)
+
+      assert(rule.convertToComet(op, stub).isEmpty)
+      assert(stub.calls == 1, s"first call should invoke getSupportLevel once, got ${stub.calls}")
+
+      assert(rule.convertToComet(op, stub).isEmpty)
+      assert(
+        stub.calls == 1,
+        s"second call should short-circuit; getSupportLevel calls = ${stub.calls}")
+    }
+  }
+
+  test("convertToComet short-circuit is per-handler (other handlers still run)") {
+    withTempView("test_data") {
+      createTestDataFrame.createOrReplaceTempView("test_data")
+
+      val sparkPlan = createSparkPlan(spark, "SELECT id FROM test_data WHERE id > 0")
+      val op = sparkPlan.collectFirst { case f: FilterExec => f }.get
+
+      val stubA = new StubUnsupportedHandlerA
+      val stubB = new StubUnsupportedHandlerB
+      val rule = CometExecRule(spark)
+
+      assert(rule.convertToComet(op, stubA).isEmpty)
+      assert(rule.convertToComet(op, stubA).isEmpty)
+      assert(
+        stubA.calls == 1,
+        s"handler A should be short-circuited on repeat, got ${stubA.calls}")
+
+      // Different handler class on the same node must still run even though A already failed.
+      assert(rule.convertToComet(op, stubB).isEmpty)
+      assert(stubB.calls == 1, s"handler B (different class) must still run, got ${stubB.calls}")
+    }
+  }
+
+  private abstract class CountingStubHandler extends CometOperatorSerde[SparkPlan] {
+    var calls: Int = 0
+    override def enabledConfig: Option[ConfigEntry[Boolean]] = None
+    override def getSupportLevel(operator: SparkPlan): SupportLevel = {
+      calls += 1
+      Unsupported(Some("stub fallback"))
+    }
+    override def convert(
+        op: SparkPlan,
+        builder: Operator.Builder,
+        childOp: Operator*): Option[OperatorOuterClass.Operator] = None
+    override def createExec(nativeOp: Operator, op: SparkPlan): CometNativeExec =
+      throw new AssertionError("createExec should not be invoked on Unsupported")
+  }
+
+  private class StubUnsupportedHandlerA extends CountingStubHandler
+  private class StubUnsupportedHandlerB extends CountingStubHandler
+
   test("CometExecRule should apply shuffle exchange transformations") {
     withTempView("test_data") {
       createTestDataFrame.createOrReplaceTempView("test_data")