experiment: remove current final agg check and see what breaks

[andygrove]

Which issue does this PR close?

Part of #2887

Rationale for this change

Comet currently falls back to Spark for final aggregates if Comet did not accelerate the partial aggregate. This is overly aggressive, and since final aggregations are often the leaf nodes in a query stage, we are unnecessarily falling back for entire stages.

What changes are included in this PR?

• Relax checks in CometHashAggregateExec so that the fallback for final aggregates only happens in cases where a Comet partial aggregate would not be compatible, rather than in all cases
• TBD we need to handle the other case - where the partial is comet but then we cannot convert the final - this requires an additional step in CometExecRule

How are these changes tested?

[codecov-commenter]

Codecov Report

❌ Patch coverage is 23.80952% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.31%. Comparing base (f09f8af) to head (a9df783).
⚠️ Report is 759 commits behind head on main.

Files with missing lines	Patch %	Lines
...n/scala/org/apache/spark/sql/comet/operators.scala	23.80%	10 Missing and 6 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #2886      +/-   ##
============================================
+ Coverage     56.12%   59.31%   +3.18%     
- Complexity      976     1374     +398     
============================================
  Files           119      167      +48     
  Lines         11743    15347    +3604     
  Branches       2251     2549     +298     
============================================
+ Hits           6591     9103    +2512     
- Misses         4012     4957     +945     
- Partials       1140     1287     +147     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[andygrove]

Spark SQL Test Failures:

These tests may need updating because Comet is now accelerating more queries.

oore-1

2025-12-12T18:08:16.7449633Z [info] *** 7 TESTS FAILED ***
2025-12-12T18:08:16.7469465Z [error] Failed: Total 9193, Failed 7, Errors 0, Passed 9186, Ignored 246, Canceled 6
2025-12-12T18:08:16.7519668Z [error] Failed tests:
2025-12-12T18:08:16.7520691Z [error] 	org.apache.spark.sql.SingleLevelAggregateHashMapSuite
2025-12-12T18:08:16.7521394Z [error] 	org.apache.spark.sql.DataFrameAggregateSuite
2025-12-12T18:08:16.7521994Z [error] 	org.apache.spark.sql.TwoLevelAggregateHashMapSuite
2025-12-12T18:08:16.7522747Z [error] 	org.apache.spark.sql.TwoLevelAggregateHashMapWithVectorizedMapSuite
2025-12-12T18:08:16.7523723Z [error] 	org.apache.spark.sql.BloomFilterAggregateQuerySuite

core-2

2025-12-12T17:48:45.4435132Z [info] *** 5 TESTS FAILED ***
2025-12-12T17:48:45.4638603Z [error] Failed: Total 3936, Failed 5, Errors 0, Passed 3746, Skipped 185, Ignored 398, Canceled 1
2025-12-12T17:48:45.4885330Z [error] Failed tests:
2025-12-12T17:48:45.4885827Z [error] 	org.apache.spark.sql.SQLQueryTestSuite
2025-12-12T17:48:45.4886480Z [error] 	org.apache.spark.sql.BloomFilterAggregateQuerySuite

core-3

2025-12-12T17:43:09.5223257Z [info] *** 8 TESTS FAILED ***
2025-12-12T17:43:09.5243848Z [error] Failed: Total 1471, Failed 8, Errors 0, Passed 1463, Ignored 38, Canceled 19
2025-12-12T17:43:09.5287616Z [error] Failed tests:
2025-12-12T17:43:09.5288557Z [error] 	org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite
2025-12-12T17:43:09.5290176Z [error] 	org.apache.spark.sql.DataFrameSuite
2025-12-12T17:43:09.5291311Z [error] 	org.apache.spark.sql.expressions.ExpressionInfoSuite

[andygrove]

Genuine failure with mixed Spark/Comet hash agg:

[info]   Cause: org.apache.comet.CometNativeException: assertion `left == right` failed: Cannot merge SparkBloomFilters with different lengths.
[info]   left: 1048588
[info]  right: 1048576
[info]         at comet::errors::init::{{closure}}(__internal__:0)
[info]         at std::panicking::panic_with_hook(__internal__:0)
[info]         at std::panicking::panic_handler::{{closure}}(__internal__:0)
[info]         at std::sys::backtrace::__rust_end_short_backtrace(__internal__:0)
[info]         at __rustc::rust_begin_unwind(__internal__:0)
[info]         at core::panicking::panic_fmt(__internal__:0)
[info]         at core::panicking::assert_failed_inner(__internal__:0)
[info]         at core::panicking::assert_failed(__internal__:0)
[info]         at datafusion_comet_spark_expr::bloom_filter::bloom_filter_agg::<impl datafusion_expr_common::accumulator::Accumulator for datafusion_comet_spark_expr::bloom_filter::spark_bloom_filter::SparkBloomFilter>::merge_batch(__internal__:0)
[info]         at <futures_util::stream::stream::fuse::Fuse<S> as futures_core::stream::Stream>::poll_next(__internal__:0)
[info]         at <datafusion_physical_plan::aggregates::no_grouping::AggregateStream as futures_core::stream::Stream>::poll_next(__internal__:0)
[info]         at <datafusion_physical_plan::projection::ProjectionStream as futures_core::stream::Stream>::poll_next(__internal__:0)
[info]         at comet::execution::jni_api::Java_org_apache_comet_Native_executePlan::{{closure}}::{{closure}}(__internal__:0)
[info]         at Java_org_apache_comet_Native_executePlan(__internal__:0)
[info]         at <unknown>(__internal__:0)