Improve performance of dropDuplicates

What is the problem the feature request solves?

Comet is 4x slower than Spark for the following query:

spark.read.parquet("/mnt/bigdata/tpcds/sf100/store_sales.parquet") \
    .repartition("ss_item_sk") \
    .dropDuplicates(["ss_item_sk", "ss_quantity"]) \
    .write.parquet("output.parquet")

If I remove the dropDuplicates call, I see similar performance between Comet and Spark.

Describe the potential solution

No response

Additional context

No response

Disabling Comet aggregation improves performance from 230s to 64s and is slightly faster than Spark at 78s.

I ran this test again with latest from main. I used this script to launch spark-shell:

$SPARK_HOME/bin/spark-shell \
  --master $SPARK_MASTER \
  --conf spark.driver.memory=4G \
  --conf spark.executor.instances=1 \
  --conf spark.executor.cores=8 \
  --conf spark.cores.max=8 \
  --conf spark.executor.memory=16g \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=16g \
  --conf spark.eventLog.enabled=false \
  --jars $COMET_JAR \
  --driver-class-path $COMET_JAR \
  --conf spark.driver.extraClassPath=$COMET_JAR \
  --conf spark.executor.extraClassPath=$COMET_JAR \
  --conf spark.plugins=org.apache.spark.CometPlugin \
  --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
  --conf spark.comet.enabled=true \
  --conf spark.comet.cast.allowIncompatible=true \
  --conf spark.comet.exec.replaceSortMergeJoin=true \
  --conf spark.comet.exec.shuffle.enabled=true \
  --conf spark.comet.exec.shuffle.fallbackToColumnar=true \
  --conf spark.comet.exec.shuffle.compression.codec=lz4 \
  --conf spark.comet.exec.shuffle.compression.level=1

I ran this query

spark.time(spark.read.parquet("/mnt/bigdata/tpcds/sf100/store_sales.parquet").repartition($"ss_item_sk").dropDuplicates(Seq("ss_item_sk", "ss_quantity")).write.parquet("output.parquet"))

With Comet: 325s
With Spark: 77s

Observations with Comet:

Scan is fast (45 seconds)
Shuffle is fast (4.3 mins)
Aggregate is slow (31.9 mins)
ColumnarToRow is slow (36.8 mins) -- this is inclusive of the aggregate time, so overhead is 4.9 mins

Observations with Spark:

Aggregate is fast (3.4 mins)
There is no ColumnarToRow

I filed #1625 for implementing a native Parquet writer so that we can avoid the C2R conversion.

Sample native plan with metrics:

25/04/08 15:06:01 INFO core/src/execution/jni_api.rs: Comet native query plan with metrics (Plan #185 Stage 7 Partition 6):
 plan creation (including CometScans fetching first batches) took 3.022557ms:
ProjectionExec: expr=[first@2 as col_0, first@3 as col_1, col_0@0 as col_2, first@4 as col_3, first@5 as col_4, first@6 as col_5, first@7 as col_6, first@8 as col_7, first@9 as col_8, first@10 as col_9, col_1@1 as col_10, first@11 as col_11, first@12 as col_12, first@13 as col_13, first@14 as col_14, first@15 as col_15, first@16 as col_16, first@17 as col_17, first@18 as col_18, first@19 as col_19, first@20 as col_20, first@21 as col_21, first@22 as col_22], metrics=[output_rows=105687, elapsed_compute=67.075µs]
  AggregateExec: mode=Final, gby=[col_0@0 as col_0, col_1@1 as col_1], aggr=[first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first], metrics=[output_rows=105687, elapsed_compute=994.964965ms, spill_count=0, spilled_bytes=0, spilled_rows=0, peak_mem_used=485980792]
    AggregateExec: mode=Partial, gby=[col_2@2 as col_0, col_10@10 as col_1], aggr=[first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first, first], metrics=[output_rows=105687, elapsed_compute=9.759887803s, spill_count=0, spilled_bytes=0, spilled_rows=0, skipped_aggregation_rows=0, peak_mem_used=485980792]
      ScanExec: source=[ShuffleQueryStage (unknown), Statistics(sizeInBytes=64.1 GiB, rowCount=2.88E+8)], schema=[col_0: Int32, col_1: Int32, col_2: Int32, col_3: Int32, col_4: Int32, col_5: Int32, col_6: Int32, col_7: Int32, col_8: Int32, col_9: Int32, col_10: Int32, col_11: Decimal128(7, 2), col_12: Decimal128(7, 2), col_13: Decimal128(7, 2), col_14: Decimal128(7, 2), col_15: Decimal128(7, 2), col_16: Decimal128(7, 2), col_17: Decimal128(7, 2), col_18: Decimal128(7, 2), col_19: Decimal128(7, 2), col_20: Decimal128(7, 2), col_21: Decimal128(7, 2), col_22: Decimal128(7, 2)], metrics=[output_rows=1508245, elapsed_compute=262.520531ms, arrow_ffi_time=12.049335ms, cast_time=1ns, jvm_fetch_time=249.312038ms]

The performance issue seems specific to the first_value aggregate function. The dropDuplicates call is equivalent to:

select ss_item_sk, ss_quantity, first_value(ss_sold_date_sk), first_value(ss_sold_time_sk), first_value(ss_customer_sk), first_value(ss_cdemo_sk), first_value(ss_hdemo_sk), first_value(ss_addr_sk), first_value(ss_store_sk), first_value(ss_promo_sk), first_value(ss_ticket_number), first_value(ss_wholesale_cost), first_value(ss_list_price), first_value(ss_sales_price), first_value(ss_ext_discount_amt), first_value(ss_ext_sales_price), first_value(ss_ext_wholesale_cost), first_value(ss_ext_list_price), first_value(ss_ext_tax), first_value(ss_coupon_amt), first_value(ss_net_paid), first_value(ss_net_paid_inc_tax), first_value(ss_net_profit) from store_sales group by ss_item_sk, ss_quantity

Using min instead of first_value is much faster.

Here is how I tested:

def bench(aggr: String) {
  val df = spark.read.parquet("/mnt/bigdata/tpcds/sf100/store_sales.parquet")
  val cols = df.schema.fieldNames.filterNot(name => Seq("ss_item_sk", "ss_quantity").contains(name))
  df.repartition($"ss_item_sk").createOrReplaceTempView("store_sales")
  val sql = s"select ss_item_sk, ss_quantity, ${cols.map(name => s"$aggr($name)").mkString(", ")} from store_sales group by ss_item_sk, ss_quantity"
  spark.sql(sql).write.parquet("output.parquet")
}

scala> spark.time(bench("first_value"))
Time taken: 320671 ms 

scala> spark.time(bench("min"))
Time taken: 64462 ms

I don't understand why there would be such a significant performance difference between first_value and min. I will try to reproduce this in DataFusion next.

There was a recent improvement in DataFusion that may help. I will test.

apache/datafusion#15266

@2010YOUY01

Running the same benchmark in a branch where I am working on the upgrade to DataFusion 47.0.0, I see that performance is greatly improved and is actually faster than Spark 🚀

scala> spark.time(bench("first_value"))
Time taken: 69429 ms

Thanks @2010YOUY01 😄

Here is a run with the original query that took 77 seconds in Spark:

scala> spark.time(spark.read.parquet("/mnt/bigdata/tpcds/sf100/store_sales.parquet").repartition($"ss_item_sk").dropDuplicates(Seq("ss_item_sk", "ss_quantity")).write.parquet("output44.parquet"))
Time taken: 62956 ms

@UBarney

Running the same benchmark in a branch where I am working on the upgrade to DataFusion 47.0.0, I see that performance is greatly improved and is actually faster than Spark 🚀
scala> spark.time(bench("first_value"))
Time taken: 69429 ms   
Thanks @2010YOUY01 😄

The credit should go to @UBarney 🚀

This should be resolved now that we have upgraded to DF 47

Navigation Menu

Repository navigation

Improve performance of `dropDuplicates` #1275

What is the problem the feature request solves?

Describe the potential solution

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Improve performance of dropDuplicates #1275

Description

What is the problem the feature request solves?

Describe the potential solution

Additional context

Activity

andygrove commented on Jan 13, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

andygrove commented on Apr 8, 2025

2010YOUY01 commented on Apr 9, 2025

andygrove commented on Apr 17, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Improve performance of `dropDuplicates` #1275