feat: Support Emit::First for SumDecimalGroupsAccumulator

[viirya]

Which issue does this PR close?

Closes #46.

Rationale for this change

If the upstream operator of HashAggregate is sorted, HashAggregate will try to emit first groups if possible. Our custom SumDecimalGroupsAccumulator doesn't support Emit::First. So if Sort is followed by HashAggregate and there is sum of decimal, the query will fail with:

        at std::backtrace::Backtrace::create(__internal__:0)                                                                                                                                                                           
        at comet::errors::init::{{closure}}(__internal__:0)                                                                                                                                                                            
        at std::panicking::rust_panic_with_hook(__internal__:0)                                                                                                                                                                        
        at std::panicking::begin_panic_handler::{{closure}}(__internal__:0)                                                                                                                                                            
        at std::sys_common::backtrace::__rust_end_short_backtrace(__internal__:0)                                                                                                                                                      
        at _rust_begin_unwind(__internal__:0)                                                                                                                                                                                          
        at core::panicking::panic_fmt(__internal__:0)                                                                                                                                                                                  
        at <comet::execution::datafusion::expressions::sum_decimal::SumDecimalGroupsAccumulator as datafusion_physical_expr::aggregate::groups_accumulator::GroupsAccumulator>::state(__internal__:0)                                  
        at datafusion_physical_plan::aggregates::row_hash::GroupedHashAggregateStream::emit(__internal__:0)                                                                                                                            
        at <datafusion_physical_plan::aggregates::row_hash::GroupedHashAggregateStream as futures_core::stream::Stream>::poll_next(__internal__:0)                                                                                     
        at <datafusion_physical_plan::aggregates::row_hash::GroupedHashAggregateStream as futures_core::stream::Stream>::poll_next(__internal__:0)                                                                                     
        at <datafusion_physical_plan::projection::ProjectionStream as futures_core::stream::Stream>::poll_next(__internal__:0)                                                                                                         
        at std::panicking::try::do_call(__internal__:0)                                                                                                                                                                                
        at _Java_org_apache_comet_Native_executePlan(__internal__:0)     

What changes are included in this PR?

How are these changes tested?

[viirya]

CI tests are passed. The failure is due to

WARNING: /Users/runner/hostedtoolcache/Java_Adopt_jdk/17.0.10-7/x64/Contents/Home/bin/java is loading libcrypto in an unsafe way

[sunchao]

@viirya I think you need to rebase this PR

[viirya]

Hmm, I think I already rebased on latest main.

[sunchao]

Hmm not sure. The same issue happened to #40 and it is OK now after the rebasing.

[viirya]

I re-triggered the pipeline.

[sunchao]

Hmm maybe the libcrypto issue is not completely fixed 😢

[viirya]

Hmm, it is actually flaky. Retriggered run is okay.

[sunchao]

LGTM. It might be better to add some details in PR description :)

[viirya]

Merged and added details in the PR description. Thanks.