[
https://issues.apache.org/jira/browse/SPARK-58411?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Josh Rosen updated SPARK-58411:
-------------------------------
Labels: correctness pull-request-available (was: pull-request-available)
> Fix bug for decorrelating Limit+Sort when partitionFields is empty
> ------------------------------------------------------------------
>
> Key: SPARK-58411
> URL: https://issues.apache.org/jira/browse/SPARK-58411
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
> Affects Versions: 4.1.0, 4.0.0, 4.2.0, 4.3.0
> Reporter: Avery Qi
> Priority: Major
> Labels: correctness, pull-request-available
> Fix For: 4.3.0
>
>
> ## Summary
> Correlated scalar/predicate subqueries of the form ORDER BY ... LIMIT k
> (optionally with OFFSET) can return wrong, non-deterministic results when the
> subquery's correlated predicate references only the outer table. During
> decorrelation, the ORDER BY is silently dropped, turning ORDER BY ... LIMIT k
> into an arbitrary LIMIT k. The row that survives then depends on physical
> execution order, so the query can return different results across
> runs/engines.
> ## Affected versions
> 4.0.0 and later (4.0.x, 4.1.x, 4.2.x, master). Introduced by SPARK-46526
> ("Support LIMIT over correlated subqueries where predicates only reference
> outer table"). Not present in 3.5.x or earlier.
> ## Root cause
> In DecorrelateInnerQuery, the Limit case peels the Sort off the subquery and
> saves its ordering into a local ordering value. When the correlation
> references only outer columns, decorrelation produces no partitioning key
> (partitionFields.isEmpty), and that branch re-emits the limit directly as
> Limit(k, newChild) (or Limit(k, Offset(m, newChild))) without ever
> re-applying the saved ordering. The ordering is only consumed by the other
> branch (the row_number() window used when there is a partitioning key). As a
> result the ordered limit degenerates into an unordered limit over the inner
> relation.
> ## Example
> SELECT t1a, t1b
> FROM t1
> WHERE t1c = (SELECT t2c
> FROM t2
> WHERE t1b < t1d -- references only outer columns
> ORDER BY t2c LIMIT 1);
> The distinct values of t2c are \{12, 16, NULL}. In Spark, ORDER BY t2c
> defaults to ASC NULLS FIRST, so the correct ordered LIMIT 1 returns NULL; the
> scalar subquery is then NULL, t1c = NULL is never true, and the query
> correctly returns no rows. With the bug, the dropped ORDER BY makes LIMIT 1
> pick an arbitrary row (e.g. a non-null 12), which can incorrectly emit a row
> and produces different results depending on runtime row order.
> ## Fix
> In the partitionFields.isEmpty branch of the Limit case, re-apply the saved
> ordering by wrapping the decorrelated child in a global Sort before the
> Limit/Offset when an ORDER BY was present. This makes ORDER BY ... LIMIT (and
> ORDER BY ... LIMIT ... OFFSET) order-preserving and deterministic, matching
> the subquery's declared semantics. The standalone Offset case (no LIMIT) is
> not affected: it decorrelates the original input with the Sort retained, so
> its ordering is already preserved.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
