From 0395dc8afeaae30ad96930504b003b316184f06b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 31 Aug 2026 08:54:36 -0600 Subject: [PATCH 1/2] chore: warn when a Compatible serde declines in convert A serde whose getSupportLevel returns Compatible is promising that convert will succeed. When convert returns None instead, the enclosing operator falls back to Spark and the JVM codegen dispatcher never gets a chance at the expression, because dispatchIfFallback is only reached from the Unsupported and Incompatible arms of exprToProtoInternal. Log a warning when that happens, so the cases show up instead of silently costing a fallback. Declines caused by a child that could not be serialized are not violations, so suppress the warning when any strict descendant already carries a fallback reason -- otherwise a single unsupported leaf would warn once per ancestor. Related to #5574. --- .github/workflows/pr_build_linux.yml | 1 + .github/workflows/pr_build_macos.yml | 1 + .../apache/comet/serde/QueryPlanSerde.scala | 54 ++++++- .../serde/CometSerdeInvariantSuite.scala | 147 ++++++++++++++++++ 4 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 081eaf76190..774111d60ed 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.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 }}] diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 074322d6a1f..d34f16459e4 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.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 diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 1268c66957d..9913c57b83d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -989,7 +989,11 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { nativeOptIn.foreach { optIn => withInfo(expr, NativeOptIn.message(exprConfName, optIn.configKey)) } - handler.convert(expr, inputs, binding) + val converted = handler.convert(expr, inputs, binding) + if (converted.isEmpty) { + warnCompatibleButDeclined(expr, handler) + } + converted } } @@ -1209,6 +1213,54 @@ 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 -- so stay quiet when any strict descendant already + * carries a fallback reason, which is how a failed child announces itself. Without that check a + * single unsupported leaf would warn once for every ancestor on the way up. + * + * 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 = { + val childDeclined = expr.children.exists { child => + child.find(_.getTagValue(CometExplainInfo.FALLBACK_REASONS).isDefined).isDefined + } + if (!childDeclined) { + // 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 diff --git a/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala b/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala new file mode 100644 index 00000000000..e70c113cb2a --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala @@ -0,0 +1,147 @@ +/* + * 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, 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 and tags it with a fallback reason. 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 when convert declines only because a child could not be converted") { + // IsNull is Compatible and does not tag itself; the decline propagates up from the child, + // which is not an invariant violation and must not warn once per ancestor. + val expr = IsNull(IsNull(TestUnregisteredExpression(Literal(1)))) + 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 tags itself but a child also failed") { + // Several serdes record their own reason when a child fails (CometCreateArray, + // CometCreateNamedStruct, CometArraysZip). A tag on the node itself must not defeat the + // child-failure check, otherwise those serdes warn on every unsupported leaf beneath them. + val child = TestUnregisteredExpression(Literal(1)) + val parent = IsNull(child) + withFallbackReason(child, "child is not supported") + withFallbackReason(parent, "unsupported arguments for parent") + val warnings = captureWarnings { + QueryPlanSerde.warnCompatibleButDeclined(parent, TestSerde) + } + assert(warnings.isEmpty, s"expected no warnings, got: $warnings") + } + + test("warns when the node is the only tagged expression in the tree") { + 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")) + } +} From 783cccdeeb64aa6fed70078ec2b9dc1657eeffb5 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 31 Aug 2026 11:07:31 -0600 Subject: [PATCH 2/2] fix: make the invariant check O(1) instead of walking the subtree The first version decided whether a decline was the node's own fault by searching the subtree for a descendant carrying a fallback reason. That runs on every convert-returns-None, so along a failing path through a large expression tree each ancestor re-walks its own subtree. This codebase demonstrably builds very large trees -- createBalancedBinaryExpr exists because And/Or chains nest deeper than protobuf's 100-level recursion limit -- so that is a real plan-time cost. Count conversion failures on the planning thread instead and compare the counter across the convert call. An unchanged counter means no nested exprToProtoInternal declined, so the decline originated in this node. Besides being O(1), this is more accurate: it also catches children the serde synthesised inside convert (a Cast wrapper, say), which are not in the expression's subtree and so were invisible to the old search. Verified equivalent on CometArrayExpressionSuite: 1 warning before and after, same expression. --- .../apache/comet/serde/QueryPlanSerde.scala | 62 ++++++++++++------- .../serde/CometSerdeInvariantSuite.scala | 38 ++++++------ 2 files changed, 61 insertions(+), 39 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 9913c57b83d..80155d8e181 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/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,8 +1008,11 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { nativeOptIn.foreach { optIn => withInfo(expr, NativeOptIn.message(exprConfName, optIn.configKey)) } + // 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) { + if (converted.isEmpty && conversionFailures.get == failuresBeforeConvert) { warnCompatibleButDeclined(expr, handler) } converted @@ -1223,9 +1245,12 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { * 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 -- so stay quiet when any strict descendant already - * carries a fallback reason, which is how a failed child announces itself. Without that check a - * single unsupported leaf would warn once for every ancestor on the way up. + * 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 @@ -1242,23 +1267,18 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { private[serde] def warnCompatibleButDeclined( expr: Expression, handler: CometExpressionSerde[_]): Unit = { - val childDeclined = expr.children.exists { child => - child.find(_.getTagValue(CometExplainInfo.FALLBACK_REASONS).isDefined).isDefined - } - if (!childDeclined) { - // 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.") - } + // 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.") } /** diff --git a/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala b/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala index e70c113cb2a..10181ed05cd 100644 --- a/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala +++ b/spark/src/test/scala/org/apache/comet/serde/CometSerdeInvariantSuite.scala @@ -26,7 +26,7 @@ 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, Expression, IsNull, Literal, Unevaluable} +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 @@ -34,8 +34,7 @@ import org.apache.comet.CometSparkSessionExtensions.withFallbackReason /** * Expression with no entry in `QueryPlanSerde.exprSerdeMap`, so `exprToProtoInternal` always - * declines it and tags it with a fallback reason. Used as the unconvertible child in the - * inherited-failure cases below. + * 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) @@ -100,31 +99,34 @@ class CometSerdeInvariantSuite extends CometTestBase { assert(warnings.head.contains("cannot resolve")) } - test("stays quiet when convert declines only because a child could not be converted") { - // IsNull is Compatible and does not tag itself; the decline propagates up from the child, - // which is not an invariant violation and must not warn once per ancestor. - val expr = IsNull(IsNull(TestUnregisteredExpression(Literal(1)))) + 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 tags itself but a child also failed") { - // Several serdes record their own reason when a child fails (CometCreateArray, - // CometCreateNamedStruct, CometArraysZip). A tag on the node itself must not defeat the - // child-failure check, otherwise those serdes warn on every unsupported leaf beneath them. - val child = TestUnregisteredExpression(Literal(1)) - val parent = IsNull(child) - withFallbackReason(child, "child is not supported") - withFallbackReason(parent, "unsupported arguments for parent") - val warnings = captureWarnings { - QueryPlanSerde.warnCompatibleButDeclined(parent, TestSerde) + 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("warns when the node is the only tagged expression in the tree") { + 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 {