andygrove commented on code in PR #4950:
URL: https://github.com/apache/datafusion-comet/pull/4950#discussion_r3889896190
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -102,6 +102,19 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
withFallbackReason(scanExec, "Full native scan disabled because
ignoreMissingFiles enabled")
}
+ // The native scan serializes the full data and partition schema (not just
the required
+ // columns), so every field type must be serializable. Types with no proto
representation
+ // (e.g. GEOMETRY / GEOGRAPHY) would otherwise crash schema serialization;
fall back instead.
+ (scanExec.relation.dataSchema.fields ++
scanExec.relation.partitionSchema.fields).foreach {
Review Comment:
Fixed in 4f23ea0d5. This was also the CI failure on this PR:
`expressions/misc/variant.sql` was the only failing test on the Spark 4.0, 4.1
and 4.2 `[expressions]` jobs, with exactly the two reasons you quoted.
Extracted the pruning `convert` was already doing into a shared
`nativeDataSchema(dataSchema, requiredSchema, resolver)` helper and validate
that in `isSupported`, so the check now covers exactly the schema the scan
serializes. The partition schema is never pruned, so it is still validated in
full. A requested actual Variant is unaffected:
`CometScanRule.isSchemaSupported` validates the projected schema and returns
before this point.
Verified `variant.sql` passes on spark-4.0, spark-4.1 and spark-4.2, and
that `geospatial_types.sql` still passes on spark-4.2, which is the GEOMETRY /
GEOGRAPHY case this check was added for.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -188,6 +188,15 @@ object CometWindowExec extends
CometOperatorSerde[WindowExec] {
val aggregateExpressions: Array[AggregateExpression] = windowExpr.flatMap
{ expr =>
expr match {
+ // Spark 4.2 allows FILTER (WHERE ...) on a window aggregate.
DataFusion window
+ // expressions have no filter, and the aggregate proto's filter field
is only honored by
+ // the native aggregate operator, so serializing this window
expression would silently
+ // evaluate the aggregate over every row of the frame and produce
wrong results.
+ case agg: AggregateExpression if agg.filter.isDefined =>
Review Comment:
Fixed in 4f23ea0d5. Reproduced it first: with a `DECIMAL(8,2)` `SUM ...
FILTER` the new case throws `Comet did not convert Window but recorded no
fallback reason on the operator or any of its expressions`, having tagged the
`MakeDecimal(...)` copy that `extractWindowExpression` builds.
Rather than threading the original through `WindowExpressionInfo`, I used
the copy-back idiom #5236 introduced for `DecimalPrecision.promote`:
`QueryPlanSerde.liftFallbackReasons` is now public, and `convert` lifts the
reasons from the rewritten window expression onto the operator's own
`windowExpression` whenever the serialized node is not part of the operator's
tree. That covers every `withFallbackReason(windowExpr, ...)` site in
`windowExprToProto`, not just the FILTER guard.
Added decimal `SUM` and `AVG` FILTER cases to `window_filter.sql`, plus
unfiltered decimal `SUM` / `AVG` controls so the fallbacks are attributable to
the filter rather than to the `DecimalAggregates` unwrapping. The file fails
without the fix and passes with it, and `CometWindowExecSuite` (52 tests)
passes on spark-3.5 and spark-4.2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
