[
https://issues.apache.org/jira/browse/SPARK-59127?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yuan Yusi updated SPARK-59127:
------------------------------
Description:
h1. AQE enabled: cached count() and collect() results disagree
h2. What is the issue?
On Apache Spark 4.2.0, a seeded non-replacement sample followed by {{cache()}}
can return inconsistent results with AQE enabled. The same cached DataFrame
reports zero rows from {{count()}} while {{collect()}} returns one row. AQE
also changes the selected sample for the fixed seed.
h2. Minimum reproduce query
{code:python}
sampled = (spark.range(1, 5, 1, 2)
.selectExpr("CAST(id AS INT) AS id")
.sort("id")
.sample(False, 0.25, 0))
cached = sampled.cache()
cached.count() # AQE on: 0
cached.collect() # AQE on: [Row(id=3)]
{code}
The range creates IDs 1 through 4 in two partitions. The sample uses fraction
0.25 and seed 0, then the sampled relation is cached before the two actions are
evaluated.
h2. Reproduction
Reproduction
Enable AQE in Spark, i.e.
{code:sql}
spark.conf.set("spark.sql.adaptive.enabled", "true")
{code}
Then run the SQL query above.
Compare the result of {{cached.collect().sizet()}} and {{cached.count()}}
h2. Test oracle
For one cached DataFrame, {{cached.count()}} must equal
{{{}cached.collect().size{}}}. With a fixed seed, the sampling must return
stable and consistent result across runs.
h2. Expected behavior
The cached relation and every action over it must agree. In particular:
{noformat}
cached_count == len(cached.collect())
{noformat}
h2. Actual behavior
{noformat}
AQE off: before=[], after=[], cached_count=0
AQE on: before=[3], after=[3], cached_count=0
{noformat}
With AQE enabled, {{cached.count()}} returns 0 while {{cached.collect()}}
returns {{{}[Row(id=3)]{}}}. The same cached result therefore has contradictory
cardinality and row contents.
h2. Plan evidence
AQE-off uses a plain {{{}Sort{}}}; AQE-on uses {{AdaptiveSparkPlan}} above the
sort. The cached collection includes an {{{}InMemoryTableScan{}}}, but the
count action still reports zero rows.
h2. Suspected root cause
{{SampleExec}} derives its seeded random stream from the input partition index.
AQE changes the post-shuffle partition layout through coalescing, and cache
materialization and count statistics are then computed at different partition
boundaries. This can leave stale or incomplete cached row-count metadata (zero)
even though a subsequent cached scan yields id 3. Disabling AQE restores
count/collect agreement.
h2. Environment
Apache Spark 4.2.0, PySpark 4.2.0, Java 17, macOS, {{{}local[2]{}}}, fraction
{{{}0.25{}}}, seed {{{}0{}}}, range {{{}(1, 5, step=1, splits=2){}}}.
was:
h1. Incorrect count() result on cached sampled DataFrame when AQE is enabled
h2. What is the issue?
On Apache Spark 4.2.0, a seeded non-replacement sample followed by {{cache()}}
can return inconsistent results with AQE enabled. The same cached DataFrame
reports zero rows from {{count()}} while {{collect()}} returns one row. AQE
also changes the selected sample for the fixed seed.
h2. Minimum reproduce query
{code:python}
sampled = (spark.range(1, 5, 1, 2)
.selectExpr("CAST(id AS INT) AS id")
.sort("id")
.sample(False, 0.25, 0))
cached = sampled.cache()
cached.count() # AQE on: 0
cached.collect() # AQE on: [Row(id=3)]
{code}
The range creates IDs 1 through 4 in two partitions. The sample uses fraction
0.25 and seed 0, then the sampled relation is cached before the two actions are
evaluated.
h2. Reproduction
Reproduction
Enable AQE in Spark, i.e.
{code:sql}
spark.conf.set("spark.sql.adaptive.enabled", "true")
{code}
Then run the SQL query above.
Compare the result of {{cached.collect().sizet()}} and {{cached.count()}}
h2. Test oracle
For one cached DataFrame, {{cached.count()}} must equal
{{{}cached.collect().size{}}}. With a fixed seed, the sampling must return
stable and consistent result across runs.
h2. Expected behavior
The cached relation and every action over it must agree. In particular:
{noformat}
cached_count == len(cached.collect())
{noformat}
h2. Actual behavior
{noformat}
AQE off: before=[], after=[], cached_count=0
AQE on: before=[3], after=[3], cached_count=0
{noformat}
With AQE enabled, {{cached.count()}} returns 0 while {{cached.collect()}}
returns {{{}[Row(id=3)]{}}}. The same cached result therefore has contradictory
cardinality and row contents.
h2. Plan evidence
AQE-off uses a plain {{{}Sort{}}}; AQE-on uses {{AdaptiveSparkPlan}} above the
sort. The cached collection includes an {{{}InMemoryTableScan{}}}, but the
count action still reports zero rows.
h2. Suspected root cause
{{SampleExec}} derives its seeded random stream from the input partition index.
AQE changes the post-shuffle partition layout through coalescing, and cache
materialization and count statistics are then computed at different partition
boundaries. This can leave stale or incomplete cached row-count metadata (zero)
even though a subsequent cached scan yields id 3. Disabling AQE restores
count/collect agreement.
h2. Environment
Apache Spark 4.2.0, PySpark 4.2.0, Java 17, macOS, {{{}local[2]{}}}, fraction
{{{}0.25{}}}, seed {{{}0{}}}, range {{{}(1, 5, step=1, splits=2){}}}.
Summary: AQE enabled: cached count() and collect() results disagree
(was: Incorrect count() result on cached sampled DataFrame when AQE is enabled
Incorrect count() result on cached sampled DataFrame when AQE is enabled)
> AQE enabled: cached count() and collect() results disagree
> ----------------------------------------------------------
>
> Key: SPARK-59127
> URL: https://issues.apache.org/jira/browse/SPARK-59127
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0
> Reporter: Yuan Yusi
> Priority: Major
> Labels: Correctness, CorrectnessBug, correctness
>
> h1. AQE enabled: cached count() and collect() results disagree
> h2. What is the issue?
> On Apache Spark 4.2.0, a seeded non-replacement sample followed by
> {{cache()}} can return inconsistent results with AQE enabled. The same cached
> DataFrame reports zero rows from {{count()}} while {{collect()}} returns one
> row. AQE also changes the selected sample for the fixed seed.
> h2. Minimum reproduce query
> {code:python}
> sampled = (spark.range(1, 5, 1, 2)
> .selectExpr("CAST(id AS INT) AS id")
> .sort("id")
> .sample(False, 0.25, 0))
> cached = sampled.cache()
> cached.count() # AQE on: 0
> cached.collect() # AQE on: [Row(id=3)]
> {code}
> The range creates IDs 1 through 4 in two partitions. The sample uses fraction
> 0.25 and seed 0, then the sampled relation is cached before the two actions
> are evaluated.
> h2. Reproduction
> Reproduction
> Enable AQE in Spark, i.e.
> {code:sql}
> spark.conf.set("spark.sql.adaptive.enabled", "true")
> {code}
> Then run the SQL query above.
> Compare the result of {{cached.collect().sizet()}} and {{cached.count()}}
> h2. Test oracle
> For one cached DataFrame, {{cached.count()}} must equal
> {{{}cached.collect().size{}}}. With a fixed seed, the sampling must return
> stable and consistent result across runs.
> h2. Expected behavior
> The cached relation and every action over it must agree. In particular:
> {noformat}
> cached_count == len(cached.collect())
> {noformat}
> h2. Actual behavior
> {noformat}
> AQE off: before=[], after=[], cached_count=0
> AQE on: before=[3], after=[3], cached_count=0
> {noformat}
> With AQE enabled, {{cached.count()}} returns 0 while {{cached.collect()}}
> returns {{{}[Row(id=3)]{}}}. The same cached result therefore has
> contradictory cardinality and row contents.
> h2. Plan evidence
> AQE-off uses a plain {{{}Sort{}}}; AQE-on uses {{AdaptiveSparkPlan}} above
> the sort. The cached collection includes an {{{}InMemoryTableScan{}}}, but
> the count action still reports zero rows.
> h2. Suspected root cause
> {{SampleExec}} derives its seeded random stream from the input partition
> index. AQE changes the post-shuffle partition layout through coalescing, and
> cache materialization and count statistics are then computed at different
> partition boundaries. This can leave stale or incomplete cached row-count
> metadata (zero) even though a subsequent cached scan yields id 3. Disabling
> AQE restores count/collect agreement.
> h2. Environment
> Apache Spark 4.2.0, PySpark 4.2.0, Java 17, macOS, {{{}local[2]{}}}, fraction
> {{{}0.25{}}}, seed {{{}0{}}}, range {{{}(1, 5, step=1, splits=2){}}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
