[SPARK-45592] AQE and InMemoryTableScanExec correctness bug

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 3.4.1, 3.5.0
• Fix Version/s: 3.4.2, 4.0.0, 3.5.1
• Component/s: SQL
• Labels: correctness pull-request-available
• Target Version/s: 3.4.2
• Assignee: Emil Ejbyfeldt
• Reporter: Emil Ejbyfeldt
• Created: 18/Oct/23 13:34
• Updated: 20/Nov/24 08:27
• Resolved: 31/Oct/23 03:23

Description

The following query should return 1000000

import org.apache.spark.storage.StorageLevel

val df = spark.range(0, 1000000, 1, 5).map(l => (l, l))
val ee = df.select($"_1".as("src"), $"_2".as("dst"))
  .persist(StorageLevel.MEMORY_AND_DISK)

ee.count()
val minNbrs1 = ee
  .groupBy("src").agg(min(col("dst")).as("min_number"))
  .persist(StorageLevel.MEMORY_AND_DISK)
val join = ee.join(minNbrs1, "src")
join.count()

but on spark 3.5.0 there is a correctness bug causing it to return `104800` or some other smaller value.

Issue Links

• relates to: SPARK-45282 Join loses records for cached datasets
• links to: GitHub Pull Request #43435
• links to: GitHub Pull Request #43729
• links to: GitHub Pull Request #43760