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..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,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 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..10181ed05cd --- /dev/null +++ b/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")) + } +}