Replace udf/udaf with Spark SQL

[Jiaweihu08]

Replace MaxWeightEstimation used in DoublePassOTreeDataAnalyzer.analyze with a simple Spark SQL code lit(1) / sum(lit(1.0) / col("normalizedWeight").

The latter is capable of leveraging vectorization, and reduce the time spent on GC.

During a testing, this change reduced the query time from 2.2h to 46mins, as shown in Stage 22 from the following image.

Test code:

val spark = SparkSession.builder().getOrCreate()

val delta = spark.read
  .format("delta")
  .load(
    "s3://qbeast-benchmarking-us-east-1/datasets/1000gb/delta/original-1tb-delta/store_sales")

val current_partitions = delta.rdd.getNumPartitions
val target_partitions = current_partitions * 4

// scalastyle:off println
println(s"Current num of partitions: $current_partitions")
println(s"Target num of partitions: $target_partitions")

val delta_repartitioned = delta.repartition(target_partitions)

val qid = QTableID("jiawei_rules")

val revision = SparkRevisionFactory
  .createNewRevision(
    qid,
    delta_repartitioned.schema,
    Map("columnsToIndex" -> "ss_item_sk,ss_ticket_number"))

val indexStatus = IndexStatus.empty(revision)

// scalastyle:on println
DoublePassOTreeDataAnalyzer.analyze(delta_repartitioned, indexStatus, false)

Cluster from EMR, 3 workers of c5.xlarge EC2 instances.

[cugni]

So much testing, for such a little change :) Great work 🔝