Yuan Yusi created SPARK-59127:
---------------------------------
Summary: Incorrect count() result on cached sampled DataFrame
when AQE is enabled Incorrect count() result on cached sampled DataFrame when
AQE is enabled
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
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){}}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
