[WIP] Add support for Hyper Log Log PLus Plus(HLL++)

[res-life]

closes ##5199

depends on

• Hyper log log plus plus(HLL++) spark-rapids-jni#2522

Description

Spark approx_count_distinct description link
Spark accepts one column(can be nested column) and a double literal relativeSD.

Depending on JNI PR:
NVIDIA/spark-rapids-jni#2522

TODO

• The NullType reduction case reports error:

struct(longs) is not supported for GPU processing yet

Perf test

// group by
import org.apache.spark.sql.functions
spark.range(10000000).repartition(5).withColumn("m", functions.expr("id % 10")).createOrReplaceTempView("tab")
spark.time(spark.sql("select m, APPROX_COUNT_DISTINCT(id) from tab group by m").show())

// reduction
spark.range(10000000).repartition(5).createOrReplaceTempView("tab")
spark.time(spark.sql("select APPROX_COUNT_DISTINCT(id) from tab ").show())

num_groups	CPU time(hot runs)	GPU time(hot runs)	speedup
10	1106ms, 1020ms, 1059ms	196ms, 208ms, 188ms	3.53x
1,000,000	5135ms, 5307ms, 5487ms	1447ms, 1565ms, 1497ms	5.38x
reduction	942ms, 1041ms, 973ms	169ms, 165ms, 180ms	5.75x

correctness

The results are identical between CPU and GPU.

Signed-off-by: Chong Gao [email protected]

[ttnghia]

Let' call by full name like GpuHyperLogLogPlusPlus to better reflect the CPU version.

[res-life]

Done

[ttnghia]

Not sure if "PlusPlus" is necessary.

Suggested change

	override val name: String = "CudfHLL"
	override val name: String = "CudfHyperLogLogPlusPlus"

[res-life]

Done.

[res-life]

Ready to review except test cases.

[revans2]

Looks good

[revans2]

nit: Using cpuAtomics for a GPU field gets to be kind of confusing. Could you please create a gpuAtomics instead?

[res-life]

Will update to support map, array and list because this is merged: NVIDIA/spark-rapids-jni#2575

[res-life]

Explain for HLLPP:
In general, HLLPP sketch is a block of memory to estimate distinct value, it contains several integer registers. The num of registers is decided by precision parameter.
num_of_registers_in_a_sketch = pow(2, precision)
e.g.: precision = 9, then num_of_registers_in_a_sketch = 2^9 = 512
Each integer register stores the number of zero bits in a hash code.
Because Spark use xxhash64 to compute hash code, thus hash code is 64 bits.
The max value of register is 64. So Refer to link

  /**
   * The number of bits that is required per register.
   *
   * This number is determined by the maximum number of leading binary zeros a hashcode can
   * produce. This is equal to the number of bits the hashcode returns. The current
   * implementation uses a 64-bit hashcode, this means 6-bits are (at most) needed to store the
   * number of leading zeros.
   */
  val REGISTER_SIZE = 6

6 bits is enough to save a register value.
Spark uses long columns to save HLLPP sketch.
e.g.: precision = 9, num_of_registers_in_a_sketch = 512
becasue of max register value is 6 bits, thus a long can hold 10 register values.
So spark uses 512/10+1 = 52 long columns to save HLLPP sketch column.
In this PR, there are some handlings the conversion:

cuDF uses Struct<long, ..., long> column to do aggregate
Convert long columns to Struct<long, ..., long> column
Convert Struct<long, ..., long> column to long columns

TODO:
Add more test cases
Support nested types: after #11859

@revans2 could you have a look first?

[res-life]

• [DONE] Add more test cases
• [DONE] Support nested types
• [DONE] Check the stack depth GPU will use does not exceed threshold

[res-life]

@ttnghia @revans2
This commit makes all test cases passed, but caused a memory leak. If have this commit, the HyperLogLogPlusPlusHostUDF(a native UDF pointer in it) is memory leaked. If do not have this commit, a core dump is caused. Refer to the following code:

case class CudfMergeHLLPP(override val dataType: DataType,
    precision: Int)

  override lazy val groupByAggregate: GroupByAggregation = {
      val hll = new HyperLogLogPlusPlusHostUDF(AggregationType.GroupByMerge, precision)
      // here hll is memory leaked; if close it, will cause core dump.
      GroupByAggregation.hostUDF(hll)
    }
}

    static final class HostUDFAggregation extends Aggregation {
       
        // this is referring a `HyperLogLogPlusPlusHostUDF`
        private final HostUDFWrapper wrapper;

        private HostUDFAggregation(HostUDFWrapper wrapper) {
            super(Kind.HOST_UDF);
            this.wrapper = wrapper;
        }

        @Override
        long createNativeInstance() {
            return Aggregation.createHostUDFAgg(wrapper.udfNativeHandle);
        }
    }

We can not close the native object in hll, because later the HostUDFAggregation.createNativeInstance will use the native object in hll.

Please discuss here to find a solution.

[res-life]

HLLPP cases passed, but have memory leak, refer to the above comment.

[res-life]

The ideal solution is that do not trigger creating a native UDF from java side. Put all the resource management into C++ layer. Not sure if it's doable.