You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2026/08/31 14:55:11 UTC
(datafusion-comet) branch andygrove/warn-compatible-convert-none created (now 0395dc8afe)
This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a change to branch andygrove/warn-compatible-convert-none
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 0395dc8afe chore: warn when a Compatible serde declines in convert
This branch includes the following new commits:
new 0395dc8afe chore: warn when a Compatible serde declines in convert
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: chore: warn when a Compatible serde declines in convert
Posted by ag...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch andygrove/warn-compatible-convert-none
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 0395dc8afeaae30ad96930504b003b316184f06b
Author: Andy Grove <ag...@apache.org>
AuthorDate: Mon Aug 31 08:54:36 2026 -0600
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 +
.../org/apache/comet/serde/QueryPlanSerde.scala | 54 +++++++-
.../comet/serde/CometSerdeInvariantSuite.scala | 147 +++++++++++++++++++++
4 files changed, 202 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml
index 081eaf7619..774111d60e 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 074322d6a1..d34f16459e 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 1268c66957..9913c57b83 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 0000000000..e70c113cb2
--- /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"))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org