chore: warn when a Compatible serde declines in convert

.github/workflows/pr_build_linux.yml

@@ -422,6 +422,7 @@ jobs:
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
               org.apache.comet.serde.CometScalarFunctionSuite
+              org.apache.comet.serde.CometSerdeInvariantSuite
               org.apache.comet.CometFallbackInvarianceSuite
       fail-fast: false
     name: ${{ matrix.profile.name }} [${{ matrix.suite.name }}]

.github/workflows/pr_build_macos.yml

@@ -238,6 +238,7 @@ jobs:
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
               org.apache.comet.serde.CometScalarFunctionSuite
+              org.apache.comet.serde.CometSerdeInvariantSuite
               org.apache.comet.CometFallbackInvarianceSuite
 
       fail-fast: false

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

@@ -917,10 +917,29 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
    *   In the case where None is returned, the expression will be tagged with the reason(s) why it
    *   is not supported.
    */
+  /**
+   * Counts expressions that failed to convert on the current planning thread. Only the
+   * before/after difference across a single [[CometExpressionSerde.convert]] call is meaningful,
+   * so the counter is never reset; see [[warnCompatibleButDeclined]] for what it is used for.
+   * Thread-local because plan compilation for different queries can run concurrently.
+   */
+  private val conversionFailures: ThreadLocal[Long] = ThreadLocal.withInitial(() => 0L)
+
   def exprToProtoInternal(
       expr: Expression,
       inputs: Seq[Attribute],
       binding: Boolean): Option[Expr] = {
+    val result = exprToProtoInternal0(expr, inputs, binding)
+    if (result.isEmpty) {
+      conversionFailures.set(conversionFailures.get + 1)
+    }
+    result
+  }
+
+  private def exprToProtoInternal0(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[Expr] = {
 
     def convert[T <: Expression](expr: T, handler: CometExpressionSerde[T]): Option[Expr] = {
       val exprConfName = handler.getExprConfigName(expr)
@@ -989,7 +1008,14 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
           nativeOptIn.foreach { optIn =>
             withInfo(expr, NativeOptIn.message(exprConfName, optIn.configKey))
           }
-          handler.convert(expr, inputs, binding)
+          // Snapshot immediately before `convert` so the comparison sees only failures that
+          // `convert` itself caused, not any recorded earlier while serializing a sibling.
+          val failuresBeforeConvert = conversionFailures.get
+          val converted = handler.convert(expr, inputs, binding)
+          if (converted.isEmpty && conversionFailures.get == failuresBeforeConvert) {
+            warnCompatibleButDeclined(expr, handler)
+          }
+          converted
       }
     }
 
@@ -1209,6 +1235,52 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     Some(ExprOuterClass.Expr.newBuilder().setScalarFunc(builder).build())
   }
 
+  /**
+   * `getSupportLevel` returning `Compatible` is a promise that `convert` will succeed, so a
+   * `None` from `convert` breaks a serde invariant. It also costs the user real performance: the
+   * `Unsupported` and `Incompatible` arms give a `CodegenDispatchFallback` serde a chance to
+   * route the expression through the JVM codegen dispatcher, whereas a decline from inside
+   * `convert` bypasses that entirely and fails the whole enclosing operator back to Spark. Such
+   * cases belong in `getSupportLevel`. See
+   * https://github.com/apache/datafusion-comet/issues/5574.
+   *
+   * Declining because a *child* could not be serialized is not a violation -- `getSupportLevel`
+   * inspects the node, not the subtree. The caller screens that out by comparing
+   * [[conversionFailures]] across the `convert` call: any nested `exprToProtoInternal` that
+   * returned `None` bumps the counter, so an unchanged counter means the decline originated in
+   * this node. That is O(1), and it also catches children the serde synthesised inside `convert`
+   * (a `Cast` wrapper, say), which are not in `expr`'s subtree and so cannot be found by
+   * inspecting it.
+   *
+   * One case warns without being a serde bug: `getSupportLevel` is not given `inputs`, so a
+   * decline that depends on them cannot be moved there. `CometAttributeReference` is the only
+   * such serde today (it declines when `bindReference` cannot resolve the attribute). The warning
+   * is still worth having there -- an unresolvable attribute is a planning anomaly worth
+   * surfacing.
+   *
+   * The warning is silent under the default configuration. Turning the codegen dispatcher off
+   * with `spark.comet.exec.scalaUDF.codegen.enabled=false` makes it fire for every serde that
+   * reports `Compatible` and then routes to `emitJvmCodegenDispatch`, since the disabled-config
+   * check lives in the dispatcher rather than in `getSupportLevel`. Those are genuine instances
+   * of this invariant violation; moving that check up is tracked on the issue above.
+   */
+  private[serde] def warnCompatibleButDeclined(
+      expr: Expression,
+      handler: CometExpressionSerde[_]): Unit = {
+    // Serdes are Scala objects, so `getSimpleName` carries a trailing `$`.
+    val serdeName = handler.getClass.getSimpleName.stripSuffix("$")
+    val reasons = expr
+      .getTagValue(CometExplainInfo.FALLBACK_REASONS)
+      .map(_.mkString("; "))
+      .getOrElse("no reason recorded")
+    logWarning(
+      s"$serdeName reported Compatible for $expr but convert() " +
+        s"returned None ($reasons), so the enclosing operator falls back to Spark. This is a " +
+        "serde invariant violation: report the case from getSupportLevel as Unsupported or " +
+        "Incompatible instead, so that a CodegenDispatchFallback serde can route it through " +
+        "the JVM codegen dispatcher.")
+  }
+
   /**
    * If `handler` is a `CodegenDispatchFallback`, run `expr` through the JVM codegen dispatcher
    * and return `Some((handler, proto))` on success; otherwise return `None`. Shared by the

spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala

@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.serde
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.logging.log4j.{Level, LogManager}
+import org.apache.logging.log4j.core.{LogEvent, Logger => Log4jLogger}
+import org.apache.logging.log4j.core.appender.AbstractAppender
+import org.apache.logging.log4j.core.config.Property
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, CreateArray, Expression, IsNull, Literal, Unevaluable}
+import org.apache.spark.sql.types.{DataType, IntegerType}
+
+import org.apache.comet.CometExplainInfo
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+
+/**
+ * Expression with no entry in `QueryPlanSerde.exprSerdeMap`, so `exprToProtoInternal` always
+ * declines it. Used as the unconvertible child in the inherited-failure cases below.
+ */
+case class TestUnregisteredExpression(child: Expression) extends Expression with Unevaluable {
+  override def children: Seq[Expression] = Seq(child)
+  override def nullable: Boolean = child.nullable
+  override def dataType: DataType = IntegerType
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): Expression =
+    copy(child = newChildren.head)
+}
+
+/** Stand-in handler; only its class name reaches the warning message. */
+private object TestSerde extends CometExpressionSerde[Expression] {
+  override def convert(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = None
+}
+
+/**
+ * A serde reporting `Compatible` from `getSupportLevel` and then returning `None` from `convert`
+ * breaks an invariant, and costs the user a Spark fallback that the JVM codegen dispatcher would
+ * otherwise have absorbed. See https://github.com/apache/datafusion-comet/issues/5574.
+ */
+class CometSerdeInvariantSuite extends CometTestBase {
+
+  private val loggerName = "org.apache.comet.serde.QueryPlanSerde"
+
+  /** Collect the WARN messages `QueryPlanSerde` emits while running `f`. */
+  private def captureWarnings(f: => Unit): Seq[String] = {
+    val captured = ArrayBuffer.empty[String]
+    val appender = new AbstractAppender("capture", null, null, true, Property.EMPTY_ARRAY) {
+      override def append(event: LogEvent): Unit = {
+        if (event.getLevel == Level.WARN) {
+          captured.synchronized(captured += event.getMessage.getFormattedMessage)
+        }
+      }
+    }
+    appender.start()
+    val logger = LogManager.getLogger(loggerName).asInstanceOf[Log4jLogger]
+    logger.addAppender(appender)
+    try {
+      f
+    } finally {
+      logger.removeAppender(appender)
+      appender.stop()
+    }
+    captured.synchronized(captured.toSeq)
+  }
+
+  private def invariantWarnings(f: => Unit): Seq[String] =
+    captureWarnings(f).filter(_.contains("serde invariant violation"))
+
+  test("warns when a Compatible serde declines in convert") {
+    // CometAttributeReference reports Compatible (IntegerType serializes fine) and then declines
+    // inside convert, because binding against an empty input list cannot resolve the attribute.
+    val attr = AttributeReference("x", IntegerType)()
+    val warnings = invariantWarnings {
+      assert(QueryPlanSerde.exprToProtoInternal(attr, Seq.empty, binding = true).isEmpty)
+    }
+    assert(warnings.size === 1, s"expected exactly one warning, got: $warnings")
+    assert(warnings.head.contains("CometAttributeReference"))
+    assert(warnings.head.contains("cannot resolve"))
+  }
+
+  test("stays quiet, once per ancestor, when only a child could not be converted") {
+    // A single unsupported leaf under a deep chain of Compatible serdes. Every ancestor's convert
+    // returns None, but none of them is at fault, so the whole tree must produce no warnings --
+    // not one per level.
+    val depth = 12
+    val expr = (1 to depth).foldLeft[Expression](TestUnregisteredExpression(Literal(1))) {
+      (child, _) => IsNull(child)
+    }
+    val warnings = invariantWarnings {
+      assert(QueryPlanSerde.exprToProtoInternal(expr, Seq.empty, binding = false).isEmpty)
+    }
+    assert(warnings.isEmpty, s"expected no warnings, got: $warnings")
+  }
+
+  test("stays quiet when the parent records its own reason but a child also failed") {
+    // CometCreateArray tags itself when a child declines. A tag on the node must not be read as
+    // evidence that the node is at fault; only the conversion-failure counter decides that.
+    val expr = CreateArray(Seq(TestUnregisteredExpression(Literal(1))))
+    val warnings = invariantWarnings {
+      assert(QueryPlanSerde.exprToProtoInternal(expr, Seq.empty, binding = false).isEmpty)
+    }
+    assert(warnings.isEmpty, s"expected no warnings, got: $warnings")
+    assert(
+      expr.getTagValue(CometExplainInfo.FALLBACK_REASONS).nonEmpty,
+      "expected CometCreateArray to have recorded its own fallback reason")
+  }
+
+  test("warning names the serde and the reason it recorded") {
+    val expr = IsNull(Literal(1))
+    withFallbackReason(expr, "declined for a reason getSupportLevel should have caught")
+    val warnings = invariantWarnings {
+      QueryPlanSerde.warnCompatibleButDeclined(expr, TestSerde)
+    }
+    assert(warnings.size === 1, s"expected exactly one warning, got: $warnings")
+    assert(warnings.head.contains("TestSerde"))
+    assert(warnings.head.contains("declined for a reason getSupportLevel should have caught"))
+  }
+
+  test("warns with a placeholder when the serde recorded no reason at all") {
+    val expr = IsNull(Literal(1))
+    assert(expr.getTagValue(CometExplainInfo.FALLBACK_REASONS).isEmpty)
+    val warnings = invariantWarnings {
+      QueryPlanSerde.warnCompatibleButDeclined(expr, TestSerde)
+    }
+    assert(warnings.size === 1, s"expected exactly one warning, got: $warnings")
+    assert(warnings.head.contains("no reason recorded"))
+  }
+}