perf: Improve count aggregate performance

[andygrove]

Which issue does this PR close?

Closes #744

Rationale for this change

For some reason, COUNT is really slow when used from Comet, but SUM is fast, so let's translate COUNT(expr) to SUM(IF(expr IS NULL, 0, 1)) until we can get to the bottom of the real issue.

edit: It turns out that Spark also implements COUNT this way, so I think this closes the issue.

Grouped HashAgg Exec: single group key (cardinality 1048576), single aggregate COUNT:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
--------------------------------------------------------------------------------------------------------------------------------------------------------------------
SQL Parquet - Spark (COUNT)                                                                    1716           1728          17          6.1         163.7       1.0X
SQL Parquet - Comet (Scan) (COUNT)                                                             1677           1680           4          6.3         159.9       1.0X
SQL Parquet - Comet (Scan, Exec) (COUNT)                                                        782            800          27         13.4          74.6       2.2X

What changes are included in this PR?

How are these changes tested?

[andygrove]

This PR results in an 11% speedup overall for TPC-DS @ 100 GB (based on a single run). Runs do vary, so this may be exaggerating the speedup. I will run with more iterations and post more results.

edit: this was run with comet.debug.enabled by mistake

[andygrove]

Average of 3 runs, main branch versus this PR. This shows a 15.5% speedup.

Command used for both runs:

$SPARK_HOME/bin/spark-submit \
    --master $SPARK_MASTER \
    --conf spark.driver.memory=8G \
    --conf spark.executor.instances=1 \
    --conf spark.executor.memory=32G \
    --conf spark.executor.cores=8 \
    --conf spark.cores.max=8 \
    --conf spark.eventLog.enabled=true \
    --jars $COMET_JAR \
    --conf spark.driver.extraClassPath=$COMET_JAR \
    --conf spark.executor.extraClassPath=$COMET_JAR \
    --conf spark.sql.extensions=org.apache.comet.CometSparkSessionExtensions \
    --conf spark.comet.enabled=true \
    --conf spark.comet.exec.enabled=true \
    --conf spark.comet.exec.all.enabled=true \
    --conf spark.comet.cast.allowIncompatible=true \
    --conf spark.comet.shuffle.enforceMode.enabled=true \
    --conf spark.comet.exec.shuffle.enabled=true \
    --conf spark.comet.exec.shuffle.mode=auto \
    --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
    tpcbench.py \
    --benchmark tpcds \
    --data /mnt/bigdata/tpcds/sf100/ \
    --queries ../../tpcds/queries-spark \
    --iterations 3

[huaxingao]

LGTM. Thanks for the PR @andygrove

[viirya]

Hmm, I think we can also do this for multiple child expressions?

[andygrove]

Thanks. I have extended this approach for the multiple argument case.

[viirya]

Looks okay. Actually it is how Spark count does internally:

 /* count = */ If(nullableChildren.map(IsNull).reduce(Or), count, count + 1L)