Improve performance on ClickBench

[Iskander14yo]

Hi!

Just made a PR to add Comet to ClickBench - one of the popular benchmarks for analytical workloads. I've decided to create an issue similar to #391. You may close it if you find it irrelevant.

I'd appreciate feedback on whether my configuration and setup are correct. I consider this important because Comet failed on one query and showed a few curious behaviors I'll outline below. Perhaps, these (and other hidden things) could be fixed with proper configuration.

My notes:

• Predictably, Comet doesn't support some expressions. That's what I got from logs:

>>> grep -P "\[COMET:" log.txt | sed -e 's/^[ \t]*//' | sort | uniq -c

     78 +-  GlobalLimit [COMET: GlobalLimit is not supported]
     18 +-  HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
    123 +-  HashAggregate [COMET: distinct aggregates are not supported]
     51 +-  Project [COMET: Unsupported cast from LongType to TimestampType with timezone Some(...) and evalMode LEGACY]
    126 +-  SortAggregate [COMET: SortAggregate is not supported]
     43 Execute CreateViewCommand [COMET: Execute CreateViewCommand is not supported]
    135 TakeOrderedAndProject [COMET: ]

Unsupported cast from LongType to TimestampType... thing is something similar to #44 but in this case another column is involved (EventTime instead of EventDate). Check this issue also for the additional info.

• Spark's local mode was used. I saw that docs suggest using standalone mode for EC2 but I didn't want to waste some extra resources on separate driver. I looked at Spark UI and seems that Comet works fine.
• Comet's cold-runs are significantly slower than hot-runs. Even compared to Spark.
• As I already mentioned, Comet failed on one query:

SELECT TraficSourceID, SearchEngineID, AdvEngineID, CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src, URL AS Dst, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 GROUP BY TraficSourceID, SearchEngineID, AdvEngineID, Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000;

with error

QueryPlanSerde: Comet native execution is disabled due to: unsupported Spark partitioning: ArrayBuffer(PageViews#1143L DESC NULLS LAST)

Caused by: org.apache.comet.CometNativeException: InternalError: Native cast invoked for unsupported cast from Utf8 to Dictionary(Int32, Utf8).

[parthchandra]

Thank you for trying to add Comet to Clickbench!

Your configuration looks ok. The distribution of memory between heap and off-heap is a little tricky and really depends on the datasets, and queries being run. You might try off heap to be 2x of on heap memory.

This ClickHouse/ClickBench#7 is definitely an issue that should be addressed in Clickbench (imho). The data is defined incorrectly as an integer and requires an explicit cast for us to interpret the integer value as a timestamp. For correctness we also need to know if the value has been adjusted to UTC.

We'll take a look at the failing query. It's likely an issue in Comet.

[parthchandra]

I can reproduce the above problem only after removing this partEventDate >= '2013-07-01' AND EventDate <= '2013-07-31' from the query. (With the clause all records are getting filtered out).
To reproduce:
Download data from https://datasets.clickhouse.com/hits_compatible/hits.parquet
Then in spark-shell -

scala> spark.read.parquet("file:///Users/parth/data/clickbench/hits.parquet").createOrReplaceTempView("hits");
scala> spark.sql("SELECT TraficSourceID, SearchEngineID, AdvEngineID, CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src, URL AS Dst, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND IsRefresh = 0 GROUP BY TraficSourceID, SearchEngineID, AdvEngineID, Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000").show()

[parthchandra]

With the EventDate clause we may have a second problem.

scala> spark.sql("SELECT TraficSourceID, SearchEngineID, AdvEngineID, CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src, URL AS Dst, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate  >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 GROUP BY TraficSourceID, SearchEngineID, AdvEngineID, Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000").show()
25/07/18 13:40:04 WARN CometExecRule: Comet cannot execute some parts of this plan natively (set spark.comet.explainFallback.enabled=false to disable this logging):
 LocalTableScan [COMET: LocalTableScan is not supported]

+--------------+--------------+-----------+---+---+---------+
|TraficSourceID|SearchEngineID|AdvEngineID|Src|Dst|PageViews|
+--------------+--------------+-----------+---+---+---------+
+--------------+--------------+-----------+---+---+---------+

[parthchandra]

The query succeeds with native_iceberg_compat and --conf spark.comet.scan.allowIncompatible=true

[Iskander14yo]

Thanks for the feedback!

On the failing query:
Appreciate the reminder, I had forgotten that Comet can use different readers. To avoid extra tuning, I’ll set spark.comet.scan.allowIncompatible = true, letting Comet pick the most suitable reader automatically. This fixes the issue and should also add flexibility, since native readers have their own limitations and we’re not concerned with Spark compatibility in this context.

On the error:
I managed to trace the issue to this expression:
CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src.
It seems that Comet falls back to Spark execution (Comet native execution is disabled due to: unsupported Spark partitioning: ArrayBuffer(PageViews#463L DESC NULLS LAST)), which leads to CometExecIterator being used to execute a plan. That plan includes a cast, and Arrow fails on it.
The open question (for me) is "why is the cast inserted here". Unfortunately, I don’t know Comet/Arrow internals well enough to effectively debug this further or suggest a proper fix on my own yet.

Also, I read your note about heap vs off-heap memory — had similar thoughts. I’ll try allocating more memory to Comet and see how it goes.

[parthchandra]

It seems that Comet falls back to Spark execution (Comet native execution is disabled due to: unsupported Spark partitioning: ArrayBuffer(PageViews#463L DESC NULLS LAST)), which leads to CometExecIterator being used to execute a plan. That plan includes a cast, and Arrow fails on it.

I did not see this message when I reproduced the error. I'll spend some time on debugging this. Logged a new issue for this: #2038

[Iskander14yo]

I did not see this message when I reproduced the error.

You can run ./benchmark.sh (to make it faster, remove all queries from queries.sql except the one that fails). The full logs will be saved in log.txt, where you should see output like:

25/07/22 09:19:14 WARN QueryPlanSerde: Comet native execution is disabled due to: unsupported Spark partitioning: ArrayBuffer(PageViews#425L DESC NULLS LAST)
25/07/22 09:19:15 WARN CometExecRule: Comet cannot execute some parts of this plan natively (set spark.comet.explainFallback.enabled=false to disable this logging):
CollectLimit
+- Project
   +-  GlobalLimit [COMET: GlobalLimit is not supported]
      +- CometSinkPlaceHolder
         +- CometExchange
            +- CometSort
               +- CometSinkPlaceHolder
                  +- CometColumnarExchange   <----- unsupported Spark partitioning*
                     +- CometHashAggregate
                        +- CometSinkPlaceHolder
                           +- CometExchange
                              +- CometHashAggregate
                                 +- CometProject
                                    +- CometFilter
                                       +- CometScanWrapper

[parthchandra]

Thanks @Iskander14yo. I'll try that. Do you know how the values for EventDate are meant to be interpreted? I tried days in unix epoch but that also filtered out all the records (or is that the expected result)?

[Iskander14yo]

@parthchandra query.py in the original PR contains the whole SparkSession configuration. Here you can see following:

df = spark.read.parquet("hits.parquet")
# Do casting before creating the view so no need to change to unreadable integer dates in SQL
df = df.withColumn("EventTime", F.col("EventTime").cast("timestamp"))
df = df.withColumn("EventDate", F.date_add(F.lit("1970-01-01"), F.col("EventDate")))
df.createOrReplaceTempView("hits")