bgiesbrecht opened a new issue, #59438:
URL: https://github.com/apache/spark/issues/59438
### What happens
A SELECT whose select list holds a subquery expression (`EXISTS`, `IN
(subquery)`, or a correlated scalar subquery) and that reads
a sorted `LIMIT` or `OFFSET` from a CTE, a FROM subquery or a view returns
rows that are not in the slice. No error is raised.
```sql
-- p(id, name, age): (1,'a',30) (2,'b',20) (3,'c',40) (4,'d',60) (5,'e',10)
(6,'f',50)
-- k(a, b): (1,2) (1,3) (1,5) (2,3)
WITH s AS (SELECT id, name FROM p ORDER BY age DESC LIMIT 3)
SELECT name, (SELECT count(*) FROM k WHERE k.a = id) AS c FROM s;
```
By `age DESC` the names are d f c a b e, so the slice is d, f, c and none of
them has a row in `k`.
- Expected: `(c,0) (d,0) (f,0)`
- Returned, with every setting at its default, 5 runs of 5: `(a,3) (b,1)
(c,0)`
The cause is in the optimizer. `PushProjectionThroughLimitAndOffset` moves a
`Project` below a `Limit` or an `Offset` when
`projectList.forall(_.deterministic)`. A project list that holds a subquery
expression passes that guard. The subquery expression is
rewritten into a join afterwards (`RewritePredicateSubquery` for EXISTS and
IN, the scalar subquery rewrite for the others), and the join
is placed under the moved `Project`, which is now under the `Limit` or
`Offset`. The optimized plan becomes
```
Offset 3
+- Project [name, exists AS e]
+- Join ExistenceJoin(exists), (a = id)
:- Project [id, name]
: +- Sort [age DESC NULLS LAST], true
: +- LogicalRDD [id, name, age]
+- Project [a]
+- LogicalRDD [a, b]
```
where the query asks for the join over the `Offset`. The `Sort` is now below
a join, and a sort-merge or shuffled-hash join re-partitions
its input by the join key, so the `Limit` or `Offset` is applied to rows in
an unrelated order. With a broadcast hash join the stream side
happens to keep its order, which hides the fault for the EXISTS form on
small tables when adaptive execution is on; it does not hide it for
the correlated scalar form, and not for tables above the broadcast
threshold. The LIMIT form also loses `TakeOrderedAndProject`: the whole input
is sorted and joined.
### Expected
The slice is computed first and the subquery expression is evaluated for the
rows of the slice, as written. Either the rule should not move a
`Project` whose list contains a subquery expression, or it should run only
after subquery expressions have been rewritten into joins.
### Observed
Spark 4.1.3, `local[1]`, `spark.sql.shuffle.partitions=2`, 5 runs a cell,
all 5 the same. "rule excluded" is
`spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.PushProjectionThroughLimitAndOffset`.
| statement (slice is `SELECT id, name FROM p ORDER BY age DESC ...`) |
expected | defaults (adaptive on) | adaptive off | adaptive on,
`autoBroadcastJoinThreshold=-1` | either, rule excluded |
|---|---|---|---|---|---|
| `WITH s AS (... LIMIT 3) SELECT name, (SELECT count(*) FROM k WHERE k.a =
id) FROM s` | c0 d0 f0 | **a3 b1 c0** | **a3 b1 c0** | **a3 b1 c0** | right |
| `WITH s AS (... OFFSET 3) SELECT name, (SELECT count(*) FROM k WHERE k.a =
id) FROM s` | a3 b1 e0 | **a3 c0 f0** | **a3 c0 f0** | **d0 e0 f0** | right |
| `WITH s AS (... LIMIT 3) SELECT name, EXISTS (SELECT 1 FROM k WHERE k.a =
id) FROM s` | cF dF fF | right | **bT dF eF** | **aT bT cF** | right |
| `WITH s AS (... OFFSET 3) SELECT name, EXISTS (SELECT 1 FROM k WHERE k.a =
id) FROM s` | aT bT eF | right | **aT cF fF** | **dF eF fF** | right |
| `SELECT name, EXISTS (...) FROM (... LIMIT 3) AS s` | cF dF fF | right |
**bT dF eF** | **aT bT cF** | right |
| `WITH s AS (... OFFSET 3) SELECT name FROM s WHERE EXISTS (...)` (control)
| a b | right | right | right | right |
| `SELECT name, EXISTS (...) FROM p ORDER BY age DESC OFFSET 3` (control) |
aT bT eF | right | right | right | right |
`IN (subquery)` and `NOT EXISTS` in the select list, and the EXISTS select
list under `DISTINCT`, an outer `ORDER BY` or an
aggregate (`count_if(EXISTS (...))`), behave like the EXISTS rows. A
correlated `max` behaves like the count rows. `MERGE` and `SHUFFLE_HASH` join
hints inside the subquery make the EXISTS rows wrong
under the default settings too; a `BROADCAST` hint hides them. Explicit
joins of the slice, and EXISTS or IN in `WHERE`, are right.
At a size above the broadcast threshold the EXISTS form is wrong with
default settings: with `p` at 200,000 rows and `k` at 2,000,000 rows (parquet),
`WITH s AS (SELECT id, name FROM p ORDER BY age DESC LIMIT 1000) SELECT
name, EXISTS (SELECT 1 FROM k WHERE k.a = id) FROM s`
returned 1,000 rows of which 4 were among the true top 1,000, in every run;
`OFFSET 199000` returned 1,000 rows of which 8 were right.
The same statements are wrong on a Databricks SQL warehouse (version
2026.39) as well, there under a broadcast hash join too: its plan shows no
`Sort` under the join.
The script below is self-contained (PySpark only, tables inline). It prints
the tallies above and the plans.
<details>
<summary>spark_slice_join_repro.py</summary>
```python
"""Apache Spark 4.1.3: a subquery expression in the select list over a
sorted LIMIT or OFFSET returns the wrong rows.
The optimizer rule PushProjectionThroughLimitAndOffset moves a Project below
a Limit or an Offset whenever the
project list is deterministic. A project list that holds a subquery
expression (EXISTS, IN, a correlated scalar
subquery) is deterministic, so it is moved too. The subquery is rewritten
into a join afterwards, and that join
then sits between the Sort and the Limit or Offset. A shuffled join does not
keep the sort order, so the slice
is taken from rows in some other order. Self-contained: pyspark only, tables
inline.
python spark_slice_join_repro.py (Java 17; PYSPARK_PYTHON set to
this Python)
p by age descending is d f c a b e. LIMIT 3 keeps d f c; OFFSET 3 keeps a b
e. k.a holds 1 1 1 2, the ids of a and b.
Each statement is run 5 times under each setting and the answers are
tallied; then the optimized and physical
plans of one statement are printed with the rule active and with it excluded.
"""
import collections
import pyspark
from pyspark.sql import SparkSession
P = [(1, "a", 30), (2, "b", 20), (3, "c", 40), (4, "d", 60), (5, "e", 10),
(6, "f", 50)]
K = [(1, 2), (1, 3), (1, 5), (2, 3)]
spark = (SparkSession.builder.master("local[1]").appName("slice-join")
.config("spark.ui.enabled",
"false").config("spark.sql.shuffle.partitions", "2").getOrCreate())
spark.sparkContext.setLogLevel("OFF")
spark.createDataFrame(P, "id BIGINT, name STRING, age
BIGINT").createOrReplaceTempView("p")
spark.createDataFrame(K, "a BIGINT, b BIGINT").createOrReplaceTempView("k")
EXISTS = "EXISTS (SELECT 1 FROM k WHERE k.a = id)"
COUNT = "(SELECT count(*) FROM k WHERE k.a = id)"
LIMIT = "SELECT id, name FROM p ORDER BY age DESC LIMIT 3"
OFFSET = "SELECT id, name FROM p ORDER BY age DESC OFFSET 3"
CASES = [
("count over LIMIT 3", f"WITH s AS ({LIMIT}) SELECT name, {COUNT} AS c
FROM s", [("c", 0), ("d", 0), ("f", 0)]),
("count over OFFSET 3", f"WITH s AS ({OFFSET}) SELECT name, {COUNT} AS c
FROM s", [("a", 3), ("b", 1), ("e", 0)]),
("EXISTS over LIMIT 3", f"WITH s AS ({LIMIT}) SELECT name, {EXISTS} AS e
FROM s", [("c", False), ("d", False), ("f", False)]),
("EXISTS over OFFSET 3", f"WITH s AS ({OFFSET}) SELECT name, {EXISTS} AS
e FROM s", [("a", True), ("b", True), ("e", False)]),
("EXISTS over LIMIT 3, FROM subquery", f"SELECT name, {EXISTS} AS e FROM
({LIMIT}) AS s", [("c", False), ("d", False), ("f", False)]),
("control: EXISTS in WHERE over OFFSET 3", f"WITH s AS ({OFFSET}) SELECT
name FROM s WHERE {EXISTS}", [("a",), ("b",)]),
("control: one SELECT, ORDER BY and OFFSET 3 last", f"SELECT name,
{EXISTS} AS e FROM p ORDER BY age DESC OFFSET 3", [("a", True), ("b", True),
("e", False)]),
]
RULE =
"org.apache.spark.sql.catalyst.optimizer.PushProjectionThroughLimitAndOffset"
SETTINGS = [
("defaults (adaptive on)", {}),
("adaptive off", {"spark.sql.adaptive.enabled": "false"}),
("adaptive on, autoBroadcastJoinThreshold=-1",
{"spark.sql.autoBroadcastJoinThreshold": "-1"}),
("adaptive off, rule excluded", {"spark.sql.adaptive.enabled": "false",
"spark.sql.optimizer.excludedRules": RULE}),
("adaptive on, autoBroadcastJoinThreshold=-1, rule excluded",
{"spark.sql.autoBroadcastJoinThreshold": "-1",
"spark.sql.optimizer.excludedRules": RULE}),
]
KEYS = ["spark.sql.adaptive.enabled",
"spark.sql.autoBroadcastJoinThreshold", "spark.sql.optimizer.excludedRules"]
RUNS = 5
def apply(conf):
for key in KEYS:
spark.conf.unset(key)
for key, value in conf.items():
spark.conf.set(key, value)
print("pyspark", pyspark.__version__, "| spark", spark.version)
for label, query, want in CASES:
print(f"\n{label}\n {query}\n expected: {want}")
for setting, conf in SETTINGS:
apply(conf)
seen = collections.Counter(str(sorted(tuple(r) for r in
spark.sql(query).collect())) for _ in range(RUNS))
verdict = "right" if set(seen) == {str(want)} else "WRONG"
print(f" {verdict:5} {setting}: " + " ; ".join(f"{n} of {RUNS}:
{rows}" for rows, n in seen.most_common()))
label, query, _ = CASES[3]
for setting, conf in (SETTINGS[1], SETTINGS[3]):
apply(conf)
df = spark.sql(query)
df.collect()
print(f"\n===== {label}; {setting} =====")
df.explain(mode="extended")
spark.stop()
```
</details>
<details>
<summary>Plans: EXISTS over OFFSET 3, adaptive off (wrong)</summary>
```
== Optimized Logical Plan ==
Offset 3
+- Project [name#1, exists#858 AS e#856]
+- Join ExistenceJoin(exists#858), (a#3L = id#0L)
:- Project [id#0L, name#1]
: +- Sort [age#2L DESC NULLS LAST], true
: +- LogicalRDD [id#0L, name#1, age#2L], false
+- Project [a#3L]
+- LogicalRDD [a#3L, b#4L], false
== Physical Plan ==
CollectLimit -1, 3
+- *(6) Project [name#1, exists#858 AS e#856]
+- *(6) SortMergeJoin [id#0L], [a#3L], ExistenceJoin(exists#858)
:- *(3) Sort [id#0L ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(id#0L, 2), ENSURE_REQUIREMENTS,
[plan_id=27123]
: +- *(2) Project [id#0L, name#1]
: +- *(2) Sort [age#2L DESC NULLS LAST], true, 0
: +- Exchange rangepartitioning(age#2L DESC NULLS LAST, 2),
ENSURE_REQUIREMENTS, [plan_id=27118]
: +- *(1) Scan ExistingRDD[id#0L,name#1,age#2L]
+- *(5) Sort [a#3L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(a#3L, 2), ENSURE_REQUIREMENTS,
[plan_id=27129]
+- *(4) Project [a#3L]
+- *(4) Scan ExistingRDD[a#3L,b#4L]
```
</details>
<details>
<summary>Plans: the same statement with the rule excluded (right)</summary>
```
== Optimized Logical Plan ==
Project [name#1, exists#862 AS e#860]
+- Join ExistenceJoin(exists#862), (a#3L = id#0L)
:- Offset 3
: +- Project [id#0L, name#1]
: +- Sort [age#2L DESC NULLS LAST], true
: +- LogicalRDD [id#0L, name#1, age#2L], false
+- Project [a#3L]
+- LogicalRDD [a#3L, b#4L], false
== Physical Plan ==
*(7) Project [name#1, exists#862 AS e#860]
+- *(7) SortMergeJoin [id#0L], [a#3L], ExistenceJoin(exists#862)
:- *(4) Sort [id#0L ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(id#0L, 2), ENSURE_REQUIREMENTS,
[plan_id=27224]
: +- *(3) GlobalLimit -1, 3
: +- *(3) Project [id#0L, name#1]
: +- *(3) Sort [age#2L DESC NULLS LAST], false, 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS,
[plan_id=27218]
: +- *(2) Project [id#0L, name#1, age#2L]
: +- *(2) Sort [age#2L DESC NULLS LAST], true, 0
: +- Exchange rangepartitioning(age#2L DESC NULLS
LAST, 2), ENSURE_REQUIREMENTS, [plan_id=27213]
: +- *(1) Scan ExistingRDD[id#0L,name#1,age#2L]
+- *(6) Sort [a#3L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(a#3L, 2), ENSURE_REQUIREMENTS,
[plan_id=27230]
+- *(5) Project [a#3L]
+- *(5) Scan ExistingRDD[a#3L,b#4L]
```
</details>
### Workaround
- Session: `SET
spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.PushProjectionThroughLimitAndOffset`
(appended to any existing value). Every statement above is right with it. On
3.4 to 4.0 the rule is named `PushProjectionThroughLimit`.
- Text, where the conf cannot be set: put an operator between the slice and
its reader that the rule does not cross, for example
`WITH s AS (SELECT * FROM (SELECT id, name FROM p ORDER BY age DESC LIMIT
3) DISTRIBUTE BY id) SELECT name, EXISTS (...) FROM s`.
This was right under every setting above and keeps `TakeOrderedAndProject`
for the LIMIT. `/*+ REPARTITION(1) */` and `/*+ COALESCE(1) */` over the slice
work as well,
unless `spark.sql.optimizer.disableHints=true`. A filter such as `WHERE
rand() >= 0`, an outer `OFFSET 0`, or a second CTE over the slice do not help.
### Versions
- Spark 4.1.3 (PySpark, local mode), Java 17, Python 3.12: LIMIT and OFFSET
forms wrong.
- Spark 4.0.1 and 3.5.6 (PySpark, local mode): the LIMIT forms are wrong the
same way and are right with `PushProjectionThroughLimit` excluded; the OFFSET
forms are right.
- By the source, the LIMIT case dates from 3.4.0 (SPARK-40501 added
`PushProjectionThroughLimit`) and the OFFSET case from 4.1.0 (SPARK-51584 added
the Offset case; SPARK-51635 merged the two rules). The rule is unchanged on
master as of this report.
- Databricks SQL 2026.39: wrong for the EXISTS and scalar count forms over
`OFFSET 3` and `LIMIT 3`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
