bgiesbrecht opened a new issue, #59434:
URL: https://github.com/apache/spark/issues/59434
**Version:** Apache Spark 4.1.3 (PySpark 4.1.3, `local[1]`, Java 17, Linux
x86_64 and macOS). Other versions not tested.
**Component:** SQL (adaptive query execution). A correctness fault: the row
count of a LIMIT is not honoured.
## Description
A query with one `LIMIT ... OFFSET` over another, over a LEFT JOIN, now and
then returns more rows than
the outer LIMIT allows. It happens only with
`spark.sql.adaptive.enabled=true` (the default). With
adaptive execution off, every run returns the right number of rows.
```sql
SELECT x FROM (
SELECT x FROM (
SELECT p.id AS x
FROM p LEFT JOIN (
SELECT p.id AS anchor FROM p INNER JOIN k ON p.id = k.a INNER JOIN p
AS q ON k.b = q.id
) o ON p.id = o.anchor
LIMIT 6 OFFSET 2
) LIMIT 3 OFFSET 1
)
```
`p` has 7 rows and `k` has 8; the join yields more than 8 rows, so the inner
slice yields 6 rows and the
outer one must yield exactly 3. There is no ORDER BY, so WHICH rows come
back may vary; the COUNT may not.
## Observed
The script below runs the query 40 times per setting and tallies the row
counts.
| machine | adaptive on | adaptive off |
|---|---|---|
| Linux x86_64, Java 17 | 3 rows: 37 runs, **5 rows: 3 runs** | 3 rows: 40
runs |
| macOS, Java 17 | 3 rows: 37 runs, **5 rows: 3 runs** | 3 rows: 40 runs |
| a third run of an earlier form of the script | 3 rows: 36, **5 rows: 4** |
3 rows: 40 |
Expected: 3 rows in every run.
## Plans
Both settings plan the same limits: a `GlobalLimit` with arguments `8, 2`
(limit 8, offset 2) under a
`CollectLimit` with arguments `4, 1`, and a `LocalLimit 8` pushed to the
left side of the outer join.
With adaptive execution on, the `GlobalLimit` reads a `ShuffleQueryStage`
and the `CollectLimit` sits in
a `ResultQueryStage`:
```
AdaptiveSparkPlan
+- == Final Plan ==
ResultQueryStage
+- CollectLimit Arguments: 4, 1
+- * GlobalLimit Arguments: 8, 2
+- ShuffleQueryStage
+- Exchange
+- * LocalLimit Arguments: 8
+- * Project
+- * SortMergeJoin LeftOuter
:- * Sort
: +- AQEShuffleRead
: +- ShuffleQueryStage
: +- Exchange
: +- * LocalLimit Arguments: 8
...
```
With it off:
```
CollectLimit Arguments: 4, 1
+- * GlobalLimit Arguments: 8, 2
+- Exchange
+- * LocalLimit Arguments: 8
+- * Project
+- * SortMergeJoin LeftOuter
...
```
Five rows is what `CollectLimit 4, 1` would give if the `GlobalLimit 8, 2`
under it had not been applied
(8 rows, less the 2 and then the 1 skipped, is 5; with it applied, 6 rows,
limit 4 offset 1, is 3). We
have not traced it further than that. The full formatted plans for both
settings are at the end of this issue.
A related form, from a real query: an inner `LIMIT 6 OFFSET 1` below a LEFT
JOIN and an aggregate, with an
`ORDER BY ... LIMIT 2` above, returned 5 or 6 rows for the LIMIT 2 in 12 of
20 runs with adaptive execution
on and 0 of 20 with it off; there the non-adaptive physical plan keeps
`TakeOrderedAndProject(limit=2)`.
## To reproduce
PySpark only; the two tables are inline; about a minute.
```python
"""Apache Spark 4.1.3: with adaptive query execution on, a LIMIT ... OFFSET
over another LIMIT ... OFFSET
over a LEFT JOIN sometimes returns more rows than its LIMIT. Self-contained:
pyspark only, tables inline.
python spark_aqe_limit_offset.py (Java 17; PYSPARK_PYTHON set to
this Python)
Expected: every run returns 3 rows (LIMIT 3). Seen: with
spark.sql.adaptive.enabled=true some runs return 5;
with it false every run returns 3. Each mode is run 40 times and the row
counts are tallied; then the
formatted physical plan of each mode is printed.
"""
import collections
import pyspark
from pyspark.sql import SparkSession
P = [(1, "a", 35, "x"), (2, "b", 15, "y"), (3, "c", 45, "x"), (4, "d", 65,
"y"),
(5, "e", 5, "x"), (6, "f", 55, "y"), (7, "g", 25, "z")]
K = [(1, 2), (1, 3), (2, 3), (3, 4), (4, 1), (5, 1), (6, 7), (6, 4)]
spark = (SparkSession.builder.master("local[1]").appName("aqe-limit-offset")
.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, g
STRING").createOrReplaceTempView("p")
spark.createDataFrame(K, "a BIGINT, b BIGINT").createOrReplaceTempView("k")
INNER = ("SELECT p.id AS x FROM p LEFT JOIN (SELECT p.id AS anchor FROM p
INNER JOIN k ON p.id = k.a "
"INNER JOIN p AS q ON k.b = q.id) o ON p.id = o.anchor")
QUERY = f"SELECT x FROM (SELECT x FROM ({INNER} LIMIT 6 OFFSET 2) LIMIT 3
OFFSET 1)"
RUNS = 40
print("pyspark", pyspark.__version__, "| spark", spark.version)
print("query:", QUERY)
for aqe in ("true", "false"):
spark.conf.set("spark.sql.adaptive.enabled", aqe)
counts = collections.Counter(len(spark.sql(QUERY).collect()) for _ in
range(RUNS))
print(f"spark.sql.adaptive.enabled={aqe}: expected 3 rows; row count:
times in {RUNS} runs = {dict(counts)}")
for aqe in ("true", "false"):
spark.conf.set("spark.sql.adaptive.enabled", aqe)
df = spark.sql(QUERY)
df.collect()
print(f"\n===== physical plan, spark.sql.adaptive.enabled={aqe} (after a
run) =====")
df.explain(mode="formatted")
spark.stop()
```
## Workaround
`spark.sql.adaptive.enabled=false` for the statement.
<details><summary>Output of the script: tallies and both formatted physical
plans</summary>
```
pyspark 4.1.3 | spark 4.1.3
query: SELECT x FROM (SELECT x FROM (SELECT p.id AS x FROM p LEFT JOIN
(SELECT p.id AS anchor FROM p INNER JOIN k ON p.id = k.a INNER JOIN p AS q ON
k.b = q.id) o ON p.id = o.anchor LIMIT 6 OFFSET 2) LIMIT 3 OFFSET 1)
spark.sql.adaptive.enabled=true: expected 3 rows; row count: times in 40
runs = {5: 3, 3: 37}
spark.sql.adaptive.enabled=false: expected 3 rows; row count: times in 40
runs = {3: 40}
===== physical plan, spark.sql.adaptive.enabled=true (after a run) =====
== Physical Plan ==
AdaptiveSparkPlan (61)
+- == Final Plan ==
ResultQueryStage (35)
+- CollectLimit (34)
+- * GlobalLimit (33)
+- ShuffleQueryStage (32), Statistics(sizeInBytes=128.0 B,
rowCount=8)
+- Exchange (31)
+- * LocalLimit (30)
+- * Project (29)
+- * SortMergeJoin LeftOuter (28)
:- * Sort (7)
: +- AQEShuffleRead (6)
: +- ShuffleQueryStage (5),
Statistics(sizeInBytes=112.0 B, rowCount=7)
: +- Exchange (4)
: +- * LocalLimit (3)
: +- * Project (2)
: +- * Scan ExistingRDD (1)
+- * Sort (27)
+- * Project (26)
+- * BroadcastHashJoin Inner BuildRight (25)
:- * Project (22)
: +- * BroadcastHashJoin Inner BuildLeft
(21)
: :- BroadcastQueryStage (15),
Statistics(sizeInBytes=1024.1 KiB, rowCount=7)
: : +- BroadcastExchange (14)
: : +- AQEShuffleRead (13)
: : +- ShuffleQueryStage (12),
Statistics(sizeInBytes=112.0 B, rowCount=7)
: : +- Exchange (11)
: : +- * Project (10)
: : +- * Filter (9)
: : +- * Scan
ExistingRDD (8)
: +- AQEShuffleRead (20)
: +- ShuffleQueryStage (19),
Statistics(sizeInBytes=192.0 B, rowCount=8)
: +- Exchange (18)
: +- * Filter (17)
: +- * Scan ExistingRDD (16)
+- BroadcastQueryStage (24),
Statistics(sizeInBytes=1024.1 KiB, rowCount=7)
+- ReusedExchange (23)
+- == Initial Plan ==
CollectLimit (60)
+- GlobalLimit (59)
+- Exchange (58)
+- LocalLimit (57)
+- Project (56)
+- SortMergeJoin LeftOuter (55)
:- Sort (37)
: +- Exchange (36)
: +- LocalLimit (3)
: +- Project (2)
: +- Scan ExistingRDD (1)
+- Sort (54)
+- Exchange (53)
+- Project (52)
+- SortMergeJoin Inner (51)
:- Sort (45)
: +- Exchange (44)
: +- Project (43)
: +- SortMergeJoin Inner (42)
: :- Sort (39)
: : +- Exchange (38)
: : +- Project (10)
: : +- Filter (9)
: : +- Scan ExistingRDD (8)
: +- Sort (41)
: +- Exchange (40)
: +- Filter (17)
: +- Scan ExistingRDD (16)
+- Sort (50)
+- Exchange (49)
+- Project (48)
+- Filter (47)
+- Scan ExistingRDD (46)
(1) Scan ExistingRDD [codegen id : 1]
Output [4]: [id#0L, name#1, age#2L, g#3]
Arguments: [id#0L, name#1, age#2L, g#3], MapPartitionsRDD[4] at
applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD,
UnknownPartitioning(0)
(2) Project [codegen id : 1]
Output [1]: [id#0L]
Input [4]: [id#0L, name#1, age#2L, g#3]
(3) LocalLimit [codegen id : 1]
Input [1]: [id#0L]
Arguments: 8
(4) Exchange
Input [1]: [id#0L]
Arguments: hashpartitioning(id#0L, 2), ENSURE_REQUIREMENTS, [plan_id=33878]
(5) ShuffleQueryStage
Output [1]: [id#0L]
Arguments: 0
(6) AQEShuffleRead
Input [1]: [id#0L]
Arguments: coalesced
(7) Sort [codegen id : 5]
Input [1]: [id#0L]
Arguments: [id#0L ASC NULLS FIRST], false, 0
(8) Scan ExistingRDD [codegen id : 2]
Output [4]: [id#808L, name#809, age#810L, g#811]
Arguments: [id#808L, name#809, age#810L, g#811], MapPartitionsRDD[4] at
applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD,
UnknownPartitioning(0)
(9) Filter [codegen id : 2]
Input [4]: [id#808L, name#809, age#810L, g#811]
Condition : isnotnull(id#808L)
(10) Project [codegen id : 2]
Output [1]: [id#808L]
Input [4]: [id#808L, name#809, age#810L, g#811]
(11) Exchange
Input [1]: [id#808L]
Arguments: hashpartitioning(id#808L, 2), ENSURE_REQUIREMENTS, [plan_id=33885]
(12) ShuffleQueryStage
Output [1]: [id#808L]
Arguments: 1
(13) AQEShuffleRead
Input [1]: [id#808L]
Arguments: local
(14) BroadcastExchange
Input [1]: [id#808L]
Arguments: HashedRelationBroadcastMode(List(input[0, bigint, true]),false),
[plan_id=34079]
(15) BroadcastQueryStage
Output [1]: [id#808L]
Arguments: 5
(16) Scan ExistingRDD [codegen id : 3]
Output [2]: [a#4L, b#5L]
Arguments: [a#4L, b#5L], MapPartitionsRDD[9] at applySchemaToPythonRDD at
NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)
(17) Filter [codegen id : 3]
Input [2]: [a#4L, b#5L]
Condition : (isnotnull(a#4L) AND isnotnull(b#5L))
(18) Exchange
Input [2]: [a#4L, b#5L]
Arguments: hashpartitioning(a#4L, 2), ENSURE_REQUIREMENTS, [plan_id=33892]
(19) ShuffleQueryStage
Output [2]: [a#4L, b#5L]
Arguments: 2
(20) AQEShuffleRead
Input [2]: [a#4L, b#5L]
Arguments: coalesced
(21) BroadcastHashJoin [codegen id : 6]
Left keys [1]: [id#808L]
Right keys [1]: [a#4L]
Join type: Inner
Join condition: None
(22) Project [codegen id : 6]
Output [2]: [id#808L, b#5L]
Input [3]: [id#808L, a#4L, b#5L]
(23) ReusedExchange [Reuses operator id: 14]
Output [1]: [id#812L]
(24) BroadcastQueryStage
Output [1]: [id#812L]
Arguments: 7
(25) BroadcastHashJoin [codegen id : 6]
Left keys [1]: [b#5L]
Right keys [1]: [id#812L]
Join type: Inner
Join condition: None
(26) Project [codegen id : 6]
Output [1]: [id#808L AS anchor#806L]
Input [3]: [id#808L, b#5L, id#812L]
(27) Sort [codegen id : 6]
Input [1]: [anchor#806L]
Arguments: [anchor#806L ASC NULLS FIRST], false, 0
(28) SortMergeJoin [codegen id : 7]
Left keys [1]: [id#0L]
Right keys [1]: [anchor#806L]
Join type: LeftOuter
Join condition: None
(29) Project [codegen id : 7]
Output [1]: [id#0L AS x#807L]
Input [2]: [id#0L, anchor#806L]
(30) LocalLimit [codegen id : 7]
Input [1]: [x#807L]
Arguments: 8
(31) Exchange
Input [1]: [x#807L]
Arguments: SinglePartition, ENSURE_REQUIREMENTS, [plan_id=34256]
(32) ShuffleQueryStage
Output [1]: [x#807L]
Arguments: 8
(33) GlobalLimit [codegen id : 8]
Input [1]: [x#807L]
Arguments: 8, 2
(34) CollectLimit
Input [1]: [x#807L]
Arguments: 4, 1
(35) ResultQueryStage
Output [1]: [x#807L]
Arguments: 9
(36) Exchange
Input [1]: [id#0L]
Arguments: hashpartitioning(id#0L, 2), ENSURE_REQUIREMENTS, [plan_id=33807]
(37) Sort
Input [1]: [id#0L]
Arguments: [id#0L ASC NULLS FIRST], false, 0
(38) Exchange
Input [1]: [id#808L]
Arguments: hashpartitioning(id#808L, 2), ENSURE_REQUIREMENTS, [plan_id=33791]
(39) Sort
Input [1]: [id#808L]
Arguments: [id#808L ASC NULLS FIRST], false, 0
(40) Exchange
Input [2]: [a#4L, b#5L]
Arguments: hashpartitioning(a#4L, 2), ENSURE_REQUIREMENTS, [plan_id=33792]
(41) Sort
Input [2]: [a#4L, b#5L]
Arguments: [a#4L ASC NULLS FIRST], false, 0
(42) SortMergeJoin
Left keys [1]: [id#808L]
Right keys [1]: [a#4L]
Join type: Inner
Join condition: None
(43) Project
Output [2]: [id#808L, b#5L]
Input [3]: [id#808L, a#4L, b#5L]
(44) Exchange
Input [2]: [id#808L, b#5L]
Arguments: hashpartitioning(b#5L, 2), ENSURE_REQUIREMENTS, [plan_id=33799]
(45) Sort
Input [2]: [id#808L, b#5L]
Arguments: [b#5L ASC NULLS FIRST], false, 0
(46) Scan ExistingRDD
Output [4]: [id#812L, name#813, age#814L, g#815]
Arguments: [id#812L, name#813, age#814L, g#815], MapPartitionsRDD[4] at
applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD,
UnknownPartitioning(0)
(47) Filter
Input [4]: [id#812L, name#813, age#814L, g#815]
Condition : isnotnull(id#812L)
(48) Project
Output [1]: [id#812L]
Input [4]: [id#812L, name#813, age#814L, g#815]
(49) Exchange
Input [1]: [id#812L]
Arguments: hashpartitioning(id#812L, 2), ENSURE_REQUIREMENTS, [plan_id=33800]
(50) Sort
Input [1]: [id#812L]
Arguments: [id#812L ASC NULLS FIRST], false, 0
(51) SortMergeJoin
Left keys [1]: [b#5L]
Right keys [1]: [id#812L]
Join type: Inner
Join condition: None
(52) Project
Output [1]: [id#808L AS anchor#806L]
Input [3]: [id#808L, b#5L, id#812L]
(53) Exchange
Input [1]: [anchor#806L]
Arguments: hashpartitioning(anchor#806L, 2), ENSURE_REQUIREMENTS,
[plan_id=33808]
(54) Sort
Input [1]: [anchor#806L]
Arguments: [anchor#806L ASC NULLS FIRST], false, 0
(55) SortMergeJoin
Left keys [1]: [id#0L]
Right keys [1]: [anchor#806L]
Join type: LeftOuter
Join condition: None
(56) Project
Output [1]: [id#0L AS x#807L]
Input [2]: [id#0L, anchor#806L]
(57) LocalLimit
Input [1]: [x#807L]
Arguments: 8
(58) Exchange
Input [1]: [x#807L]
Arguments: SinglePartition, ENSURE_REQUIREMENTS, [plan_id=33815]
(59) GlobalLimit
Input [1]: [x#807L]
Arguments: 8, 2
(60) CollectLimit
Input [1]: [x#807L]
Arguments: 4, 1
(61) AdaptiveSparkPlan
Output [1]: [x#807L]
Arguments: isFinalPlan=true
===== physical plan, spark.sql.adaptive.enabled=false (after a run) =====
== Physical Plan ==
CollectLimit (30)
+- * GlobalLimit (29)
+- Exchange (28)
+- * LocalLimit (27)
+- * Project (26)
+- * SortMergeJoin LeftOuter (25)
:- * Sort (5)
: +- Exchange (4)
: +- * LocalLimit (3)
: +- * Project (2)
: +- * Scan ExistingRDD (1)
+- * Sort (24)
+- Exchange (23)
+- * Project (22)
+- * SortMergeJoin Inner (21)
:- * Sort (18)
: +- Exchange (17)
: +- * Project (16)
: +- * SortMergeJoin Inner (15)
: :- * Sort (10)
: : +- Exchange (9)
: : +- * Project (8)
: : +- * Filter (7)
: : +- * Scan ExistingRDD (6)
: +- * Sort (14)
: +- Exchange (13)
: +- * Filter (12)
: +- * Scan ExistingRDD (11)
+- * Sort (20)
+- ReusedExchange (19)
(1) Scan ExistingRDD [codegen id : 1]
Output [4]: [id#0L, name#1, age#2L, g#3]
Arguments: [id#0L, name#1, age#2L, g#3], MapPartitionsRDD[4] at
applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD,
UnknownPartitioning(0)
(2) Project [codegen id : 1]
Output [1]: [id#0L]
Input [4]: [id#0L, name#1, age#2L, g#3]
(3) LocalLimit [codegen id : 1]
Input [1]: [id#0L]
Arguments: 8
(4) Exchange
Input [1]: [id#0L]
Arguments: hashpartitioning(id#0L, 2), ENSURE_REQUIREMENTS, [plan_id=34403]
(5) Sort [codegen id : 2]
Input [1]: [id#0L]
Arguments: [id#0L ASC NULLS FIRST], false, 0
(6) Scan ExistingRDD [codegen id : 3]
Output [4]: [id#818L, name#819, age#820L, g#821]
Arguments: [id#818L, name#819, age#820L, g#821], MapPartitionsRDD[4] at
applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD,
UnknownPartitioning(0)
(7) Filter [codegen id : 3]
Input [4]: [id#818L, name#819, age#820L, g#821]
Condition : isnotnull(id#818L)
(8) Project [codegen id : 3]
Output [1]: [id#818L]
Input [4]: [id#818L, name#819, age#820L, g#821]
(9) Exchange
Input [1]: [id#818L]
Arguments: hashpartitioning(id#818L, 2), ENSURE_REQUIREMENTS, [plan_id=34409]
(10) Sort [codegen id : 4]
Input [1]: [id#818L]
Arguments: [id#818L ASC NULLS FIRST], false, 0
(11) Scan ExistingRDD [codegen id : 5]
Output [2]: [a#4L, b#5L]
Arguments: [a#4L, b#5L], MapPartitionsRDD[9] at applySchemaToPythonRDD at
NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)
(12) Filter [codegen id : 5]
Input [2]: [a#4L, b#5L]
Condition : (isnotnull(a#4L) AND isnotnull(b#5L))
(13) Exchange
Input [2]: [a#4L, b#5L]
Arguments: hashpartitioning(a#4L, 2), ENSURE_REQUIREMENTS, [plan_id=34415]
(14) Sort [codegen id : 6]
Input [2]: [a#4L, b#5L]
Arguments: [a#4L ASC NULLS FIRST], false, 0
(15) SortMergeJoin [codegen id : 7]
Left keys [1]: [id#818L]
Right keys [1]: [a#4L]
Join type: Inner
Join condition: None
(16) Project [codegen id : 7]
Output [2]: [id#818L, b#5L]
Input [3]: [id#818L, a#4L, b#5L]
(17) Exchange
Input [2]: [id#818L, b#5L]
Arguments: hashpartitioning(b#5L, 2), ENSURE_REQUIREMENTS, [plan_id=34423]
(18) Sort [codegen id : 8]
Input [2]: [id#818L, b#5L]
Arguments: [b#5L ASC NULLS FIRST], false, 0
(19) ReusedExchange [Reuses operator id: 9]
Output [1]: [id#822L]
(20) Sort [codegen id : 10]
Input [1]: [id#822L]
Arguments: [id#822L ASC NULLS FIRST], false, 0
(21) SortMergeJoin [codegen id : 11]
Left keys [1]: [b#5L]
Right keys [1]: [id#822L]
Join type: Inner
Join condition: None
(22) Project [codegen id : 11]
Output [1]: [id#818L AS anchor#816L]
Input [3]: [id#818L, b#5L, id#822L]
(23) Exchange
Input [1]: [anchor#816L]
Arguments: hashpartitioning(anchor#816L, 2), ENSURE_REQUIREMENTS,
[plan_id=34509]
(24) Sort [codegen id : 12]
Input [1]: [anchor#816L]
Arguments: [anchor#816L ASC NULLS FIRST], false, 0
(25) SortMergeJoin [codegen id : 13]
Left keys [1]: [id#0L]
Right keys [1]: [anchor#816L]
Join type: LeftOuter
Join condition: None
(26) Project [codegen id : 13]
Output [1]: [id#0L AS x#817L]
Input [2]: [id#0L, anchor#816L]
(27) LocalLimit [codegen id : 13]
Input [1]: [x#817L]
Arguments: 8
(28) Exchange
Input [1]: [x#817L]
Arguments: SinglePartition, ENSURE_REQUIREMENTS, [plan_id=34535]
(29) GlobalLimit [codegen id : 14]
Input [1]: [x#817L]
Arguments: 8, 2
(30) CollectLimit
Input [1]: [x#817L]
Arguments: 4, 1
```
</details>
--
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]
