[EPIC] Improve performance of TPC-H queries

[andygrove]

What is the problem the feature request solves?

This epic is for tracking progress on improving performance of Comet with our benchmarks derived from TPC-H.

Current status (September 2024)

• Comet is 1.6x faster than Spark
• Comet is not as fast as other DataFusion subprojects yet
• All of these DataFusion subprojects are performing similar native execution, which indicates that there is room to improve on Comet's current performance

Features needed to support all queries natively

We do not run all queries fully natively yet due to these missing features:

• Support sort merge join with a join condition #398 (q17, q19, q20, q21)
• Comet doesn't support Spark BroadcastHashJoinExec if it is null-aware anti-join #457 (q16)
• Add support for bloom_filter_agg #846 (q5, q7, q8, q20, q21)

Planned features that could help in general

• [EPIC] Improve shuffle performance #1123
• Add option to replace SortMergeJoin with ShuffleHashJoin #1006
• Implement native version of ColumnarToRow #708
• Implement native RowToColumnarExec
• Improve performance of Spark-compatible decimal aggregates #951
• Support RangePartitioning with native shuffle #458
• Implement Common Subexpression Elimination optimizer rule #942
• Avoid unpacking dictionaries for inputs to SortExec #963
• Optimize filters to remove redundant IsNotNull checks #938

Issues that affect multiple queries

• Scans are sometimes slower due to dictionary encoding or decoding, and it may be better if we can defer this until later in the query, but this is not really possible at the moment because DataFusion requires that all batches with a stream have the same physical type, so we cannot match Utf and Dictionary for example
• CometExchange is sometimes slower than Spark's exchange even though it reads and writes less data.

Per-Query Tracking

Most of these queries are already faster with Comet enabled. Here are notes on areas where performance could potentially be improved.

• q1
  • Implement Common Subexpression Elimination optimizer rule #942
• q2
  • Some scans are slower, partly due to dictionary unpacking cost
  • Some exchanges are slower (could this also be due to premature unpacking of dictionaries?)
• q3
• q4
• q5
  • Add support for bloom_filter_agg #846
• q6
• q7
  • Add support for bloom_filter_agg #846
• q8
  • Add support for bloom_filter_agg #846
• q9
  • 82% of the time is in SortExec + SortMergeJoinExec. Ballista uses a HashJoinExec.
• q10
• q11
• q12
• q13
• q14
  • lineitem scans take 2x longer in Comet, but this is offset by avoiding an expensive C2R. The time for native decoding in Comet is longer than the entire scan in Spark.
  • Implement Common Subexpression Elimination optimizer rule #942
• q15
• q16
  • Comet doesn't support Spark BroadcastHashJoinExec if it is null-aware anti-join #457
• q17
  • Support sort merge join with a join condition #398
• q18
• q19
  • Support sort merge join with a join condition #398
• q20
  • Add support for bloom_filter_agg #846
  • Support sort merge join with a join condition #398
• q21
  • Add support for bloom_filter_agg #846
  • Support sort merge join with a join condition #398
• q22

[viirya]

BroadcastExchange should be supported, I think. We have CometBroadcastExchange.

We don't need to support AQEShuffleRead. It is a shuffle reader wrapper in Spark. It calls wrapped shuffle's execute or executeColumnar depending on it is columnar or not.

[viirya]

We don't need to support Execute CreateViewCommand too. It is a command exec operator.

[viirya]

Also CommandResult, which is only used to hold data from a command. CommandResult and Execute CreateViewCommand are not query execution operators.

[andygrove]

Also CommandResult, which is only used to hold data from a command. CommandResult and Execute CreateViewCommand are not query execution operators.

Thanks. I saw those from the CREATE VIEW in q15 but I see from the Spark UI that the SELECT part of this query is already fully native. I have removed those from the list.

[andygrove]

BroadcastExchange should be supported, I think. We have CometBroadcastExchange.

BroadcastExchange is not supported is the information that Comet provides for q8. I think part of this epic will be making these messages more informative.

[viirya]

For Sort merge join with a join condition, I added the support to DataFusion for a while but we've not incorporated the feature in Comet yet. I opened #398 to track it and I will work on it once #250 is merged and #248 is done.

[viirya]

BroadcastExchange is not supported is the information that Comet provides for q8. I think part of this epic will be making these messages more informative.

I will take a look at q8 and see why it is not enabled there.

[andygrove]

I will take a look at q8 and see why it is not enabled there.

The error BroadcastExchange is not supported really means BroadcastExchange is not supported because the child operators are not supported

[viirya]

Please disable spark.comet.exec.broadcast.enabled which should not be used in normal query: #408 (comment)

[mbutrovich]

I ran TPC-H locally and profiled the sole executor with 4 CPU cores allocated to it. One thing I noticed is that update_comet_metric is taking 3.2% of the time. Within Native_executePlan it accounts for ~7-8% of an individual worker's CPU time in Comet.

I want to look at the granularity that these operations occur at, and see if we can coalesce metrics on the native side and maybe ship more at once to reduce the JNI overhead. I want to add more metrics to Comet to understand where we're spending time, but the overhead is going to add up.