[BUG] Incorrect output from averages with filters in partial only mode

[kuhushukla]

Describe the bug
When using filters on averages with conf set to run only the partial mode on the gpu, the result can be different from the CPU.

Steps/Code to reproduce bug
Adding the following test to hash_aggregate_test.py shows the failure behaviour.

@ignore_order
@allow_non_gpu(
    'HashAggregateExec', 'AggregateExpression',
    'AttributeReference', 'Alias', 'Sum', 'Count', 'Max', 'Min', 'Average', 'Cast',
    'KnownFloatingPointNormalized', 'NormalizeNaNAndZero', 'GreaterThan', 'Literal', 'If',
    'EqualTo', 'First', 'SortAggregateExec', 'Coalesce')
@pytest.mark.parametrize('data_gen', get_params(_init_list_no_nans,
    marked_params=params_markers_for_avg_sum), ids=idfn)
def test_hash_multiple_filters_fail(data_gen):
    df = with_cpu_session(
        lambda spark : gen_df(spark, data_gen, length=100))
    df.createOrReplaceTempView("hash_agg_table")
    assert_gpu_and_cpu_are_equal_collect(
        lambda spark: spark.sql(
            'select avg(b) filter (where b > 20)' +
            'from hash_agg_table group by a'),
        conf=_no_nans_float_conf_partial)

Expected behavior
cpu result = 266969243.5
gpu result = None

[kuhushukla]

This is a tough one given the current design. Pushing the filter down to the agg function like avg() doesn't work and if we want to special case it for average in GpuAggregateExpressions wrapper, we would have to assume a few things or reach into the original aggregate function to check what it is and if it is an average. Appreciate any comments on if disabling is the way forward, or special casing would be OK.

[abellina]

Could you add a little more info on what is hard about? And also the special case solution, what is the issue with that approach?

I am a little confused in how having a partial cpu hash agg would impact filter.

[kuhushukla]

SUre.
The issue pertains to filters but only because of nulls being passed down to the CPU. Let me elaborate and hopefully we can find a way.
Input is
column names = (a,b)
values (1, 15)
(1, null)
say we are calculating avg(b) and number of partitions is 2 so one partition gets(1,15)another gets (1, null)
Averages in partial mode output (sum,count) columns. In this case for the second columnar batch we will have (0.0,0) where both the bits have validity vector set to false which means they are nulls. The CPU however is expecting that nulls in partial mode have come out as (0.0,0) i.e. valid zero values. If you remeber this is what we had even with defaultValues for GpuAverage.
"Overall the answer computed becomes null instead of 15 in this case"
The reason why filter fails is because we use GpuCaseWhen which will emit nulls if filter doesnt match and we need to convert this null to a valid 0 before we send it down to the CPU. Since WrappedAggFunction is the only place we have access to the filter we would need to detect that the origAggFunction is an Average and then make sure we dont pass down nulls. That is what I mean by special casing.

I tried an approach where I was forcing the knowledge that we have a filter or basically something that can cause nulls down to GpuAverage via GpuDecalarativeAggregate as that seems like the right place anyway to detect filters and do the GpuCaseWhen but I had some issues with the count aggregate not working correctly as I had the filter set to none during convertToGpu and by the time I set it , it is too late. Average cannot access the filter straight up so detecting that we are doing a filter in the wrapped class on an average specifically is the only idea I have right now.