fix: address review feedback on mixed partial/final aggregate guard

- Restore `convert` scaladoc in `CometAggregateExpressionSerde` that was
  displaced when `supportsMixedPartialFinal` was added
- Require `aggregateExpressions.nonEmpty` in `findPartialAggInPlan` so
  intermediate distinct-elimination stages (empty agg, group-by only)
  are not incorrectly tagged as the Partial to disable
- Document that `canFinalAggregateBeConverted` mirrors the predicate
  checks in `CometBaseAggregate.doConvert` and must be kept in sync

Author: andygrove

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -647,6 +647,11 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
    * operator enablement, grouping expressions, aggregate expressions, and result expressions.
    * Intentionally skips the sparkFinalMode / child-native checks since those depend on
    * transformation state.
+   *
+   * WARNING: this intentionally mirrors the predicate checks in `CometBaseAggregate.doConvert`
+   * (operators.scala). Any change to the convertibility rules there must be reflected here or
+   * this tagging pass will drift and either crash (missed tag) or over-disable (spurious tag). A
+   * shared predicate helper would be preferable.
    */
   private def canFinalAggregateBeConverted(agg: BaseAggregateExec): Boolean = {
     val handler = allExecs.get(agg.getClass)
@@ -690,11 +695,15 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
 
   /**
    * Search the child subtree for the first Partial-mode aggregate, traversing through exchanges
-   * and AQE stages.
+   * and AQE stages. Requires `aggregateExpressions.nonEmpty` so that intermediate distinct stages
+   * (group-by-only aggregates with empty aggregateExpressions, where `.forall` vacuously matches)
+   * are not mistaken for the partial we want to tag.
    */
   private def findPartialAggInPlan(plan: SparkPlan): Option[BaseAggregateExec] = {
     plan.collectFirst {
-      case agg: BaseAggregateExec if agg.aggregateExpressions.forall(e => e.mode == Partial) =>
+      case agg: BaseAggregateExec
+          if agg.aggregateExpressions.nonEmpty &&
+            agg.aggregateExpressions.forall(e => e.mode == Partial) =>
         Some(agg)
       case a: AQEShuffleReadExec => findPartialAggInPlan(a.child)
       case s: ShuffleQueryStageExec => findPartialAggInPlan(s.plan)

spark/src/main/scala/org/apache/comet/serde/CometAggregateExpressionSerde.scala

@@ -49,6 +49,14 @@ trait CometAggregateExpressionSerde[T <: AggregateFunction] {
    */
   def getSupportLevel(expr: T): SupportLevel = Compatible(None)
 
+  /**
+   * Whether this aggregate's intermediate buffer format is compatible between Spark and Comet,
+   * making it safe to run the Partial in one engine and the Final in the other. Aggregates with
+   * simple single-value buffers (MIN, MAX, COUNT, bitwise) are safe; those with complex or
+   * differently-encoded buffers (AVG, SUM with decimals, CollectSet, Variance) are not.
+   */
+  def supportsMixedPartialFinal: Boolean = false
+
   /**
    * Convert a Spark expression into a protocol buffer representation that can be passed into
    * native code.
@@ -68,14 +76,6 @@ trait CometAggregateExpressionSerde[T <: AggregateFunction] {
    *   case it is expected that the input expression will have been tagged with reasons why it
    *   could not be converted.
    */
-  /**
-   * Whether this aggregate's intermediate buffer format is compatible between Spark and Comet,
-   * making it safe to run the Partial in one engine and the Final in the other. Aggregates with
-   * simple single-value buffers (MIN, MAX, COUNT, bitwise) are safe; those with complex or
-   * differently-encoded buffers (AVG, SUM with decimals, CollectSet, Variance) are not.
-   */
-  def supportsMixedPartialFinal: Boolean = false
-
   def convert(
       aggExpr: AggregateExpression,
       expr: T,