You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "bgiesbrecht (via GitHub)" <gi...@apache.org> on 2026/10/13 15:49:01 UTC
[I] Wrong result: LIMIT/OFFSET over LIMIT/OFFSET over a LEFT JOIN returns more rows than its LIMIT when adaptive query execution is on [spark]
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: reviews-unsubscribe@spark.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [I] Wrong result: LIMIT/OFFSET over LIMIT/OFFSET over a LEFT JOIN returns more rows than its LIMIT when adaptive query execution is on [spark]
Posted by "AbarnaaSree (via GitHub)" <gi...@apache.org>.
AbarnaaSree commented on issue #59434:
URL: https://github.com/apache/spark/issues/59434#issuecomment-6250472388
Hi, I’d like to help investigate this AQE correctness issue. I can work on a focused regression test that reproduces the nested `LIMIT/OFFSET` behavior over a `LEFT JOIN`, verify the result count with adaptive execution enabled and disabled, and inspect how `GlobalLimit`, `CollectLimit`, and shuffle query stages interact in the adaptive physical plan. I’ll also check whether the issue can be reproduced reliably across repeated executions before proposing a fix. Would this investigation be useful to the maintainers?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [I] Wrong result: LIMIT/OFFSET over LIMIT/OFFSET over a LEFT JOIN returns more rows than its LIMIT when adaptive query execution is on [spark]
Posted by "Raviston6296 (via GitHub) via reviews" <re...@spark.apache.org>.
Raviston6296 commented on issue #59434:
URL: https://github.com/apache/spark/issues/59434#issuecomment-6570979658
This is the same root cause as SPARK-57956 (#57347): during AQE re-optimization, `LogicalQueryStage.maxRows` used a cost estimate of a not-yet-materialized stage as a hard upper bound, so `EliminateLimits` could drop a LIMIT. That fix is in `master`, `branch-4.x` and `branch-4.3`, but not in `branch-4.2` or `branch-4.1`.
I reproduced the query from this issue on `branch-4.2` and `branch-4.1` (RDD-backed tables, `local[1]`, `spark.sql.shuffle.partitions=2`, 200 runs with AQE on): 19 and 18 runs returned 5 rows. With SPARK-57956 backported, all 200 runs return 3 rows on both branches. Current master also returns 3 rows in every run.
Backports: #60017 (`branch-4.2`) and #60018 (`branch-4.1`).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org