Peter Toth created SPARK-60106:
----------------------------------
Summary: PlanMerger can build join pairs that no merged subplan
needs
Key: SPARK-60106
URL: https://issues.apache.org/jira/browse/SPARK-60106
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
With filter propagation through joins on, `PlanMerger` can merge two subplans
where one plan filters one join child and the other plan filters the other
child. The merged join then reads both children unfiltered and also builds the
pairs that neither subplan needs. The answer is right, but the join work can be
many times larger than the two original joins together.
It only needs this config, which is off by default:
*
{{spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled=true}}
Repro, with a non-nullable join key:
{code:scala}
spark.range(20000).selectExpr("id", "coalesce(id % 100, 0L) AS k", "id AS
v").createOrReplaceTempView("l")
spark.range(20000).selectExpr("id", "coalesce(id % 100, 0L) AS k", "id AS
v").createOrReplaceTempView("r")
sql("""
SELECT
(SELECT sum(l.v) FROM l JOIN r ON l.k = r.k WHERE l.id < 100),
(SELECT max(r.v) FROM l JOIN r ON l.k = r.k WHERE r.id < 100)
""")
{code}
The two subqueries alone produce 20,000 join rows each. Merged, the single join
produces 4,000,000.
If each plan's filter keeps a share p and q of its side, the merged join does
about 1/(p+q) times the work of the two original joins, so the merge only pays
off for filters that keep a large part of the data. Both plans have a differing
filter in this case, which elsewhere in `PlanMerger` already counts as
symmetric filter propagation and needs
{{spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
