test: move aggregate function coverage from CometAggregateSuite to Comet SQL tests #6631

Part of #6615.

What / Why

This issue moves aggregate function coverage from exec/CometAggregateSuite to Comet SQL tests under sql-tests/expressions/aggregate/. It also covers the HLL sketch tests in CometExpressionSuite. Aggregate operator behavior that only Scala can observe stays where it is. Static analysis on main at 3bc2faa found the following:

None of these tests is fully covered by an existing fixture today. The fixtures are thinner in three ways:

1. String group keys only. Every grouped fixture for sum, count, min/max, avg, bit_agg, stddev, variance, covariance, corr, first_last and collect_* groups by a string key. The Scala tests group by tinyint, smallint, int, bigint, float, double, decimal, date, timestamp, binary and VARCHAR keys. DataFusion uses a different group-values implementation for a single primitive key than for a single string key.
2. Tolerance where Scala compares exactly. The fixtures use tolerance=1e-6 for avg over ints, grouped var/stddev, and the corr NaN/NULL cases. Tolerance also skips NaN (test: query tolerance= in Comet SQL tests passes when either side is NaN #6616).
3. Whole areas have no fixture:
   • try_sum, try_avg, bool_and / every;
   • sum/avg under ANSI, and decimal SUM overflow at maximum precision;
   • spark.sql.legacy.statisticalAggregate;
   • strictFloatingPoint min/max;
   • GROUP BY without aggregate functions, and SELECT DISTINCT;
   • min/max over float, decimal, date, timestamp or boolean;
   • aggregates under spark.comet.shuffle.mode=jvm.

How checkSparkAnswerAndNumOfAggregates(q, n) maps. The helper is checkSparkAnswer (answer only) plus a count of CometHashAggregateExec. When n is 2 (or 4 for COUNT(DISTINCT)) on a fully native plan, a plain query is equal or stronger. Counts of 1 or 0 under spark.comet.shuffle.enabled=false or a disabled partial/final mode test mixed-engine behavior; those parts stay in Scala.

Batch shape. Under local[5], an INSERT ... VALUES of a few rows writes one file per row, so each partial aggregate sees one row. For ordered or multi-row input use INSERT INTO t SELECT /*+ COALESCE(1) */ ... or range(0, n, 1, 1). The Scala ORDER BY views become subqueries; the Sort survives under first/last because those aggregates are order-sensitive.

Work

The work splits into three PR-sized parts. Line numbers refer to CometAggregateSuite unless marked.

Part 1: grouping keys, types, DISTINCT, batches

• group_by_numeric_keys.sql (new)
  • Covers L856, L982, L999, L1012, L1037, L1102, L1123, L1147, L1619, L1636, L1657 (the convertible part) and L1746.
  • Use int and bigint keys with real NULLs; see the note on L1102-L1147 below.
  • SUM/COUNT/MIN/MAX/AVG grouped and ungrouped, with exact avg.
  • Result expressions such as MIN(v)+2, SUM(v+1), k+2 and sum(v + g3).
  • Add shuffle-mode and dictionary ConfigMatrix lines.
• group_by_floating_point.sql (new): L276 (group part), L1175, L1190, L2212, L2281.
  • Float and double keys with NaN, float('-0.0') / double('-0.0') and NULL.
  • Mostly singleton groups, so SUM/AVG compare exactly.
• min_max_strict_floating_point.sql (new): L276, with spark.comet.exec.strictFloatingPoint=true. Two expect_fallback queries per min/max over float/double zeros, one per reason.
• min_max_datetime.sql (new): L1303, -- ConfigMatrix: spark.sql.parquet.outputTimestampType=TIMESTAMP_MICROS,TIMESTAMP_MILLIS.
• distinct.sql (new): L686, L699, L840, L2346.
  • Multi-group count(DISTINCT) over a literal.
  • Multi-column count(DISTINCT) with a NULL in the tuple.
  • GROUP BY without aggregate functions; SELECT DISTINCT; repeated distinct aggregates.
  • Shuffle-mode ConfigMatrix.
• group_by_multi_batch.sql (new): L1320, L1434, L1506.
  • -- ConfigMatrix: spark.comet.batchSize=128,1024,10001.
  • Deterministic keys at cardinality 1, 100 and 10000, including negatives.
• first_last_multi_batch.sql (new): L1402, L1477, L1571. first/last with IGNORE NULLS over a sorted subquery, and 8192 groups.
• group_by_all_types.sql (new): L1539, L1588. Keys of all 14 primitive types, with batch-size, dictionary and outputTimestampType matrices.
• collect_grouped_batches.sql (new): L191, with spark.comet.batchSize=128.
• collect_list.sql
  • L157: a struct with a non-nullable field via coalesce, and the two-stage collect_list -> named_struct -> collect_list shape.
  • L215: collect_list / collect_set of CAST(NULL AS ARRAY<STRUCT<...>>).
• bool_and.sql (new): L2563, covering bool_and / every and min/max over boolean.
• bit_agg.sql: L2588. bit_or / bit_xor over tinyint and smallint, grouped by a smallint key, under jvm shuffle.
• first_last.sql: L2525, plus -- ConfigMatrix: spark.comet.shuffle.mode=native,jvm.
• partial_merge.sql
  • L1347: sum(string) with count(DISTINCT string) over 3 files.
  • L1388: exact query for avg(int)+count(DISTINCT), sum(1)+sum(DISTINCT) and avg(DISTINCT)+sum.
• group_by_map_fallback.sql (new): L1075, L3351, with MinSparkVersion: 4.1 and spark.comet.exec.localTableScan.enabled=true.
• hash_aggregate_pruned_output.sql (new): L3379.

Part 2: sum/avg (decimal, ANSI, try) and statistical aggregates

• sum_avg_decimal.sql (new): L309, L949, L966, L1209, L1223, L1242, L1263 (part), L1818, L2234, L2301 (part).
  • Columns: decimal(5,2), decimal(18,10) and decimal(38,37) with NULL rows; tinyint/smallint/int/bigint keys.
  • MIN/MAX/SUM/AVG, including decimal(38,37) overflow to NULL.
• sum_decimal_max_precision.sql (new): L1862, L2127, L2152, with an ANSI matrix.
  • decimal(38,38) rows 0.6, 0.6, -0.6 written as one ordered file via range(0, n, 1, 1), to keep the running-sum order.
• sum_decimal_max_precision_legacy.sql (new), plus ANSI twins and per-codegen-config siblings: the non-ANSI halves of L1885, L1907, L1934 and L2152, plus L1959, L2023 and L2093.
  • L1959 fans out to about 8 files: whole-stage codegen off, maxFields=1, and NO_CODEGEN with a 3.5+/3.4 version split.
  • L1934's error class changes at 4.0 (.WITH_SUGGESTION).
• sum_ansi.sql (new, ANSI matrix): L2926, L2939, L2984, L3003, the non-overflow parts of L3063 and L3139, and the sum part of L3286.
• sum_ansi_overflow.sql (new, ANSI on): the expect_error(ARITHMETIC_OVERFLOW) halves of L3063, L3118, L3139 and L3198. The Scala check (both engines throw, both messages contain the class) is exactly what expect_error does.
• try_sum.sql (new, ANSI matrix): L2955, L2968, L3022, L3040, L3220, L3260, L3267, L3274.
• avg_ansi.sql (new, ANSI matrix; also holds try_avg): L2188, L2843, L2879, and the avg part of L3286.
• sum.sql
  • L1028: grouped bigint overflow.
  • L1467: a VARCHAR(3) group key.
  • The non-ANSI halves of L3063, L3118, L3139 and L3198. Use a /*+ REPARTITION(2) */ hint for .repartition(2).
• avg.sql (L1050, part): AVG(int) grouped by a string key, compared exactly.
• stddev.sql / variance.sql (L671, L2762, L2801)
  • Add -- ConfigMatrix: spark.sql.legacy.statisticalAggregate=true,false, a shuffle-mode matrix and a dictionary matrix.
  • Add a one-row float('Infinity') case.
  • Add var_pop(col5). The Scala test meant to cover it but runs var_samp(col5) instead.
• covariance.sql, corr.sql (L2665, L2737)
  • The nine datasets, with legacy, shuffle and dictionary matrices.
  • Use exact query for NULL and NaN results; use tolerance with ORDER BY only for finite results.
• hll_sketch.sql (new, MinSparkVersion: 4.0): CometExpressionSuite L4044, L4069 and L4175.
  • Wrap each estimate in the test's own bound, e.g. abs(est - 700) / 700.0 <= 0.05. A plain query then asserts native execution and the 5% bound without comparing estimates bit for bit.

Part 3: operator-level tests that happen to be expressible (optional)

These tests assert only answers or fallback reasons, and every config they need is a SQL conf, so they can move. They carry no expression coverage, so migrating them is optional.

• mixed_engine_spark_partial.sql: L718, L731. A VALUES source keeps the Partial in Spark.
• mixed_engine_partial_disabled.sql: L800.
• approx_percentile_distinct.sql: L340, with a matrix over spark.comet.testing.aggregate.partialMode.enabled / finalMode.enabled.
• skip_partial_aggregation.sql: L2388.
• correlated_subquery_aggregate.sql: L874, the TPC-H Q17 shape.
• collect_distinct_fallback.sql: L240.

Blocked

• test: add a Comet SQL test error mode that checks native execution and error-class parity #6617: the ANSI halves of L1885, L1907, L1934 and L2152, and CometExpressionSuite L4097 (hll_union_agg lgConfigK). Each asserts native execution plus error-class, SQLSTATE and exception-class parity.
• test: let Comet SQL tests keep ConstantFolding, exclude optimizer rules and change configs mid-file #6618:
  • L911 and L930 need EliminateSorts excluded, so the Sort below the SUM survives.
  • CometExpressionSuite L4132 needs NullPropagation excluded.

Stays in Scala

• Metrics: L63, L105, L2422, L2449.
  • L63 and L105 are ignored with "TODO: To be addressed after DF 55 migration" and no link. Re-enable the two ignored CometAggregateSuite metric tests after the DataFusion 55 peak_mem_used change #5703 tracks peak_mem_used, so link it from the ignore.
• Mixed-engine buffer safety and plan shape: L359, L388, L455, L512, L540, L580, L746, L786, L823, L2268. These check unsafe-partial tags, Comet aggregate counts of 0 or 1, CometFilterExec presence and materialized AQE stages.
  • The shuffle-off halves of L1050, L1263 and L2301, and the java_method plan check in L1657, stay for the same reason.
• Plan canonicalization and exchange reuse: L1698.
• A helper unit test with no query: L1767.
• CometFuzzAggregateSuite (all 9). It is the fuzz layer: it sweeps whatever schema the generator produces and picks up new types automatically. The fixtures should still close the gaps it exposes. No fixture asserts native count(DISTINCT) for boolean, tinyint, smallint, float, decimal(36,18), date, timestamp, timestamp_ntz or binary.

Problems found along the way

Vacuous or misleading Scala tests

• The "null" key tests have no NULLs. L1102, L1123 and L1147 build keys with null.asInstanceOf[Int], which is 0 in Scala. Their expected rows are really Row(0, ...), and "with all nulls" aggregates zeros.
• L949 no longer runs Comet's decimal SUM. With Comet shuffle off the Final is in Spark, and CometSum.supportsNativePartialToSparkFinal is false for decimal, so the Partial is restored to Spark.
• L1347 sets spark.comet.exec.shuffle.fallbackToColumnar, which is not defined anywhere in spark/src/main.
• L686, "count with aggregation filter", has no FILTER clause.
• L2762 never tests var_pop over double.
• L3286's nullability assertion is vacuous, because df.schema comes from Spark's analyzed plan.
• L1320's view ORDER BY is dropped. EliminateSorts removes it below an order-insensitive aggregate.

Fixture problems

• aggregate_filter.sql links to the placeholder issues/XXXX, and lines 44-45 and 60-61 are the same query.
• min_max.sql's only ungrouped query mixes in string min/max, so it is expect_fallback(SortAggregate is not supported). Native ungrouped min/max over int or double is never asserted.
• "Multi-batch" comments are wrong. first_last.sql ("large group (multi-batch)") and partial_merge.sql ("large, multi-batch input") write 1000 rows as one file, below the 8192 default batch size, so both run as a single batch.
• NaN and infinity under tolerance. corr.sql test_corr_nan, avg.sql test_avg_special and sum.sql's Infinity rows compare these results only under tolerance, which skips them (test: query tolerance= in Comet SQL tests passes when either side is NaN #6616).
• collect_list.sql has groups mixing 0.0 and -0.0 under sort_array, which is order-sensitive.
• Possibly stale config. partial_merge.sql and L2301 set spark.comet.expression.Cast.allowIncompatible=true, but CometCast.canCastFromDouble now reports DecimalType as Compatible. This was not verified by running.

Done when

• Every test in the table below is accounted for. Either a fixture covers it (the same inputs, configs and assertion strength, or stronger), the fixture that already covers it is cited, or it stays in Scala for the reason given above.
• Each converted Scala test is deleted in the PR that adds its coverage. A suite left empty is deleted and removed from the suite lists in .github/workflows/pr_build_linux.yml and pr_build_macos.yml.
• The new and changed fixtures pass on every Spark profile. Apply run-all-spark-profiles, because pull requests run only the default profile.