Investigate nested TPC-H q21 slowdown: Comet 37% slower than Spark at SF1000

[comphead]

Summary

On a nested-schema benchmark derived from TPC-H (SF1000, Iceberg), q21 is the one query where Comet is clearly slower than Spark. The mean over 3 iterations is 518.4 s for Comet versus 377.6 s for Spark (+37.3%, 0.73x), and the iteration ranges do not overlap. Comet is faster than Spark on most of the other queries in this benchmark, so this looks specific to q21. q21 is also the longest query, at 45% of Comet's total runtime across the 22 queries (24% for Spark).

Results

Engine	Iteration 1 (s)	Iteration 2 (s)	Iteration 3 (s)	Mean (s)
Spark	371.732	382.426	378.716	377.625
Comet	478.485	520.237	556.420	518.381

Observations

• Comet's iteration times rise from one iteration to the next (478.5, 520.2, 556.4 s) while Spark's stay flat. The cause is not known. Memory pressure, spilling or native memory growth across iterations are possibilities, but this is only a hypothesis.
• The query has a correlated EXISTS and a correlated NOT EXISTS with an inequality predicate on l_suppkey. Spark rewrites these into semi and anti joins with a join filter.
• Earlier upstream issues mention this query: [BUG]: SortMergeJoin filtered LeftAnti fails on TPCH Q21 #861 (closed, filtered LeftAnti sort merge join failing on TPC-H q21) and perf: allocation accounting wrapper costs ~5% on TPC-H Q21 from thread-local lookups in the dlopened library #6165 (closed, allocation accounting overhead of about 5% on TPC-H q21). It is not known whether the build used here includes those fixes.
• The query uses explode over array<struct> columns. Iceberg scan falls back to Spark on IS NULL/IS NOT NULL over list/map columns (stale complex-type check) #5731 (closed) describes Iceberg scans falling back to Spark for IS NULL / IS NOT NULL predicates on list columns, which Spark pushes down for explode. It is not known whether that applies here.

Expected behavior

Comet should be at least as fast as Spark on this query.

Query

-- using default substitutions

select
	supplier_data.s_name,
	count(*) as numwait
from
	(select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier,
	(select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as l1,
	(select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders,
	(select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
where
	s_suppkey = l1.l_suppkey
	and o_orderkey = l1.l_orderkey
	and orders_data.o_orderstatus = 'F'
	and l1.lineitem_data.l_receiptdate > l1.lineitem_data.l_commitdate
	and exists (
		select
			*
		from
			lineitem l2
		where
			l2.l_orderkey = l1.l_orderkey
			and l2.l_suppkey <> l1.l_suppkey
	)
	and not exists (
		select
			*
		from
			(select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as l3
		where
			l3.l_orderkey = l1.l_orderkey
			and l3.l_suppkey <> l1.l_suppkey
			and l3.lineitem_data.l_receiptdate > l3.lineitem_data.l_commitdate
	)
	and s_nationkey = n_nationkey
	and nation_data.n_name = 'SAUDI ARABIA'
group by
	supplier_data.s_name
order by
	numwait desc,
	supplier_data.s_name
limit 100

Schema of the tables used

supplier
  s_suppkey BIGINT, s_nationkey BIGINT,
  supplier_data ARRAY<STRUCT<s_name STRING, s_address STRING, s_phone STRING, s_acctbal DECIMAL(12,2), s_comment STRING>>

lineitem (partitioned by l_shipdate)
  l_orderkey BIGINT, l_partkey BIGINT, l_suppkey BIGINT, l_shipdate DATE,
  lineitem_data ARRAY<STRUCT<l_linenumber INT, l_quantity DECIMAL(12,2), l_extendedprice DECIMAL(12,2),
    l_discount DECIMAL(12,2), l_tax DECIMAL(12,2), l_returnflag STRING, l_linestatus STRING,
    l_commitdate DATE, l_receiptdate DATE, l_shipinstruct STRING, l_shipmode STRING, l_comment STRING>>

orders (partitioned by o_orderdate)
  o_orderkey BIGINT, o_custkey BIGINT, o_orderdate DATE,
  orders_data ARRAY<STRUCT<o_orderstatus STRING, o_totalprice DECIMAL(12,2), o_orderpriority STRING,
    o_clerk STRING, o_shippriority INT, o_comment STRING>>

nation
  n_nationkey BIGINT, n_regionkey BIGINT,
  nation_data ARRAY<STRUCT<n_name STRING, n_comment STRING>>

Setup

• Benchmark: a nested-schema variant derived from TPC-H at scale factor 1000. Each table keeps its key columns (and partition columns) at the top level and stores every other column in one array<struct<...>> column named <table>_data. The queries are the TPC-H queries rewritten to use explode(<table>_data) in subqueries.
• Tables: Iceberg 1.5.0 (downstream build) with Parquet data files, zstd compression and 512 MB row groups. lineitem is partitioned by l_shipdate, orders by o_orderdate and part by p_brand.
• Cluster: Spark 3.4.3 (downstream build, Scala 2.13) on Kubernetes with 16-core amd64 executors. The Comet build is a downstream build and has not been checked against upstream main yet.
• Recorded configs (both runs): spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager, spark.comet.exec.shuffle.enabled=true, spark.comet.exec.shuffle.compression.codec=lz4, spark.comet.expression.allowIncompatible=true, spark.comet.explainFallback.enabled=true, spark.memory.fraction=0.6 and spark.memory.storageFraction=0.2.
• Method: a Spark baseline run (Comet not active) and a Comet run. Each run is one Spark application that ran all 22 queries in order, with 3 consecutive iterations per query. Times are per-iteration query execution times in seconds as recorded by the benchmark harness. There is a single run per engine.

Not verified yet

• Not reproduced on upstream main.
• No physical plans, Spark UI metrics or fallback reasons are attached yet. spark.comet.explainFallback.enabled=true was set for the runs, so fallback reasons should be available (not yet reviewed).
• Executor count, executor memory and off-heap sizing are not recorded by the benchmark harness and are not listed here.
• The harness does not record how Comet was switched off in the Spark baseline run. Both runs list the Comet shuffle manager in their recorded configs.

[dwsmith1983]

@comphead do you have the spark logs? Anything strange in them?

[comphead]

@comphead do you have the spark logs? Anything strange in them?

I feel it is different finding because it appeared on nested TPCH, not a flat one. And I'm currently investigating mem usage and event logs to identify a reason

[dwsmith1983]

@comphead I looked at q21 in the code and with a small benchmark of its two correlated subquery joins.

Spark turns the EXISTS and NOT EXISTS into LeftSemi and LeftAnti sort-merge joins with the l_suppkey <> filter, and Comet runs both natively by default (spark.comet.exec.sortMergeJoinWithJoinFilter.enabled). DataFusion evaluates that filter one inner row at a time, so on lineitem-shaped key groups the join takes 8 to 10 times as long as the same join without the filter. Each small key group is also charged a full input batch of memory, so under memory pressure the join can spill once per group. Numbers are in apache/datafusion#25923. My rough estimate puts this at tens of seconds of the gap at SF1000, not all of it, and it doesn't explain the times rising from one iteration to the next.

These would help narrow down the rest:

• the Comet commit or version the downstream build is based on, since Iceberg scan falls back to Spark on IS NULL/IS NOT NULL over list/map columns (stale complex-type check) #5731 (Iceberg scans falling back on the IS NOT NULL that explode pushes down) and fix: return a native plan's memory before its Spark task ends #6261 (native plans keeping memory after an early stop) were fixed after 1.0.0
• the q21 physical plan from both runs and the explainFallback output for q21
• spill counts and join time on the two CometSortMergeJoin nodes for each iteration
• executor memory settings, especially spark.memory.offHeap.enabled and its size, and any lost or OOMKilled executors
• a run with spark.comet.exec.sortMergeJoinWithJoinFilter.enabled=false, which sends only those two joins back to Spark
• whether other queries also get slower across their three iterations

[comphead]

Some datapoints, Spark4 Iceberg TPCH1000

Plain:
Spark 104s
Comet 62s

Nested:
Spark 367s
Comet 380s

[comphead]

Observation Comet has uncompressed spill by default, and with compression enabled

 spark.comet.datafusion.execution.spill_compression=lz4_frame

The exec time is 390s

[comphead]

The reason was indeed uncompressed spilling and comet datafusion param propagation disabled by default.

Correct conf is

--conf spark.comet.datafusion.execution.spill_compression=lz4_frame
--conf spark.comet.exec.respectDataFusionConfigs=true

Comet nested 250s

@andygrove WDYT should we enable params by default or document in Tuning Guide?

@sunchao you might be interested