From 78d438d4f8b6559f821f394b1fdda2da85b3844c Mon Sep 17 00:00:00 2001 From: peterxcli Date: Mon, 29 Jun 2026 04:17:22 +0800 Subject: [PATCH] Report unsupported scalar serde cases in support level --- .../comet/serde/contraintExpressions.scala | 17 +++-- .../apache/comet/serde/namedExpressions.scala | 76 ++++++++++--------- .../org/apache/comet/serde/unixtime.scala | 16 ++-- .../apache/comet/CometExpressionSuite.scala | 30 +++++++- 4 files changed, 87 insertions(+), 52 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala index 5ffcae2b2d..a8cbc548f9 100644 --- a/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala @@ -22,7 +22,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, KnownFloatingPointNormalized} import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, serializeDataType} object CometKnownFloatingPointNormalized @@ -33,7 +32,13 @@ object CometKnownFloatingPointNormalized override def getSupportLevel(expr: KnownFloatingPointNormalized): SupportLevel = { expr.child match { - case _: NormalizeNaNAndZero => Compatible() + case normalized: NormalizeNaNAndZero => + val wrapped = normalized.child + if (serializeDataType(wrapped.dataType).isDefined) { + Compatible() + } else { + Unsupported(Some(s"Unsupported datatype ${wrapped.dataType}")) + } case _ => Unsupported( Some( @@ -48,17 +53,13 @@ object CometKnownFloatingPointNormalized val wrapped = expr.child.asInstanceOf[NormalizeNaNAndZero].child - val dataType = serializeDataType(wrapped.dataType) - if (dataType.isEmpty) { - withFallbackReason(wrapped, s"Unsupported datatype ${wrapped.dataType}") - return None - } + val dataType = serializeDataType(wrapped.dataType).get val ex = exprToProtoInternal(wrapped, inputs, binding) val optExpr = ex.map { child => val builder = ExprOuterClass.NormalizeNaNAndZero .newBuilder() .setChild(child) - .setDatatype(dataType.get) + .setDatatype(dataType) ExprOuterClass.Expr.newBuilder().setNormalizeNanAndZero(builder).build() } optExprWithFallbackReason(optExpr, expr, wrapped) diff --git a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala index b778f2ea87..fb910ec3ac 100644 --- a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala @@ -38,53 +38,55 @@ object CometAlias extends CometExpressionSerde[Alias] { } object CometAttributeReference extends CometExpressionSerde[AttributeReference] { + override def getSupportLevel(attr: AttributeReference): SupportLevel = { + if (serializeDataType(attr.dataType).isDefined) { + Compatible() + } else { + Unsupported(Some(s"unsupported datatype: ${attr.dataType}")) + } + } + override def convert( attr: AttributeReference, inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { - val dataType = serializeDataType(attr.dataType) + val dataType = serializeDataType(attr.dataType).get - if (dataType.isDefined) { - if (binding) { - // Spark may produce unresolvable attributes in some cases, - // for example https://github.com/apache/datafusion-comet/issues/925. - // So, we allow the binding to fail. - val boundRef: Any = BindReferences - .bindReference(attr, inputs, allowFailures = true) + if (binding) { + // Spark may produce unresolvable attributes in some cases, + // for example https://github.com/apache/datafusion-comet/issues/925. + // So, we allow the binding to fail. + val boundRef: Any = BindReferences + .bindReference(attr, inputs, allowFailures = true) - if (boundRef.isInstanceOf[AttributeReference]) { - withFallbackReason(attr, s"cannot resolve $attr among ${inputs.mkString(", ")}") - return None - } + if (boundRef.isInstanceOf[AttributeReference]) { + withFallbackReason(attr, s"cannot resolve $attr among ${inputs.mkString(", ")}") + return None + } - val boundExpr = ExprOuterClass.BoundReference - .newBuilder() - .setIndex(boundRef.asInstanceOf[BoundReference].ordinal) - .setDatatype(dataType.get) - .build() + val boundExpr = ExprOuterClass.BoundReference + .newBuilder() + .setIndex(boundRef.asInstanceOf[BoundReference].ordinal) + .setDatatype(dataType) + .build() - Some( - ExprOuterClass.Expr - .newBuilder() - .setBound(boundExpr) - .build()) - } else { - val unboundRef = ExprOuterClass.UnboundReference + Some( + ExprOuterClass.Expr .newBuilder() - .setName(attr.name) - .setDatatype(dataType.get) - .build() - - Some( - ExprOuterClass.Expr - .newBuilder() - .setUnbound(unboundRef) - .build()) - } + .setBound(boundExpr) + .build()) } else { - withFallbackReason(attr, s"unsupported datatype: ${attr.dataType}") - None - } + val unboundRef = ExprOuterClass.UnboundReference + .newBuilder() + .setName(attr.name) + .setDatatype(dataType) + .build() + Some( + ExprOuterClass.Expr + .newBuilder() + .setUnbound(unboundRef) + .build()) + } } } diff --git a/spark/src/main/scala/org/apache/comet/serde/unixtime.scala b/spark/src/main/scala/org/apache/comet/serde/unixtime.scala index 1a3a4ca677..7bcc03c38f 100644 --- a/spark/src/main/scala/org/apache/comet/serde/unixtime.scala +++ b/spark/src/main/scala/org/apache/comet/serde/unixtime.scala @@ -22,19 +22,28 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, FromUnixTime, Literal} import org.apache.spark.sql.catalyst.util.TimestampFormatter +import org.apache.comet.CometConf import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} // TODO: DataFusion supports only -8334601211038 <= sec <= 8210266876799 // https://github.com/apache/datafusion/issues/16594 object CometFromUnixTime extends CometExpressionSerde[FromUnixTime] with CodegenDispatchFallback { + private val unsupportedFormatReason = "Datetime pattern format is unsupported" override def getIncompatibleReasons(): Seq[String] = Seq( "Only supports the default datetime format pattern `yyyy-MM-dd HH:mm:ss`." + " DataFusion's valid timestamp range differs from Spark" + " (https://github.com/apache/datafusion/issues/16594)") - override def getSupportLevel(expr: FromUnixTime): SupportLevel = Incompatible(None) + override def getSupportLevel(expr: FromUnixTime): SupportLevel = { + if (expr.format == Literal(TimestampFormatter.defaultPattern()) || + !CometConf.isExprAllowIncompat(getExprConfigName(expr))) { + Incompatible(None) + } else { + Unsupported(Some(unsupportedFormatReason)) + } + } override def convert( expr: FromUnixTime, @@ -48,10 +57,7 @@ object CometFromUnixTime extends CometExpressionSerde[FromUnixTime] with Codegen val formatExpr = exprToProtoInternal(Literal("%Y-%m-%d %H:%M:%S"), inputs, binding) val timeZone = exprToProtoInternal(Literal(expr.timeZoneId.orNull), inputs, binding) - if (expr.format != Literal(TimestampFormatter.defaultPattern)) { - withFallbackReason(expr, "Datetime pattern format is unsupported") - None - } else if (secExpr.isDefined && formatExpr.isDefined) { + if (secExpr.isDefined && formatExpr.isDefined) { val timestampExpr = scalarFunctionExprToProto("from_unixtime", Seq(secExpr, timeZone): _*) val optExpr = scalarFunctionExprToProto("to_char", Seq(timestampExpr, formatExpr): _*) diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index abe8291c63..5ec15befea 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -25,8 +25,9 @@ import scala.util.Random import org.apache.hadoop.fs.Path import org.apache.spark.sql.{Column, CometTestBase, DataFrame, Row} -import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, FromUnixTime, Literal, StructsToJson, TruncDate, TruncTimestamp} -import org.apache.spark.sql.catalyst.optimizer.SimplifyExtractValueOps +import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Cast, FromUnixTime, KnownFloatingPointNormalized, Literal, StructsToJson, TruncDate, TruncTimestamp} +import org.apache.spark.sql.catalyst.optimizer.{NormalizeNaNAndZero, SimplifyExtractValueOps} +import org.apache.spark.sql.catalyst.util.TimestampFormatter import org.apache.spark.sql.comet.CometProjectExec import org.apache.spark.sql.execution.{ProjectExec, SparkPlan} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper @@ -36,6 +37,7 @@ import org.apache.spark.sql.internal.SQLConf.SESSION_LOCAL_TIMEZONE import org.apache.spark.sql.types._ import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, isSpark41Plus, isSpark42Plus} +import org.apache.comet.serde.{CometAttributeReference, CometFromUnixTime, CometKnownFloatingPointNormalized, Incompatible, Unsupported} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator} class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -1047,6 +1049,30 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("misc scalar serdes report static unsupported cases via getSupportLevel") { + val intervalAttr = AttributeReference("interval_attr", CalendarIntervalType)() + + assert( + CometAttributeReference.getSupportLevel(intervalAttr) == + Unsupported(Some("unsupported datatype: CalendarIntervalType"))) + assert( + CometFromUnixTime.getSupportLevel( + FromUnixTime(Literal(0L), Literal(TimestampFormatter.defaultPattern()), None)) == + Incompatible(None)) + assert( + CometFromUnixTime.getSupportLevel(FromUnixTime(Literal(0L), Literal("yyyy-MM-dd"))) == + Incompatible(None)) + withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[FromUnixTime]) -> "true") { + assert( + CometFromUnixTime.getSupportLevel(FromUnixTime(Literal(0L), Literal("yyyy-MM-dd"))) == + Unsupported(Some("Datetime pattern format is unsupported"))) + } + assert( + CometKnownFloatingPointNormalized.getSupportLevel( + KnownFloatingPointNormalized(NormalizeNaNAndZero(intervalAttr))) == + Unsupported(Some("Unsupported datatype CalendarIntervalType"))) + } + test("rlike with non-scalar pattern runs via codegen dispatcher") { val table = "rlike_non_scalar" withTable(table) {