Spark executor OOM while joining very small dataset (non-zero exit code 143)

I am doing left outer join on a small data set(total 41 mb). My executor memory is 10GB. When i am not using any in memory persistence on executor the job works fine. But when i use Cache() or persist(StorageLevel.MEMORY_AND_DISK()) the job consistently fails with below exception (but works when i use persist(StorageLevel.DISK_ONLY()))

# java.lang.OutOfMemoryError: Java heap space
# -XX:OnOutOfMemoryError="kill %p"
#   Executing /bin/sh -c "kill 42447"...
23/06/14 19:38:57 WARN TaskMemoryManager: Failed to allocate a page (64102862 bytes), try again.
23/06/14 19:38:57 INFO TaskMemoryManager: Memory used in task 4
23/06/14 19:38:57 INFO TaskMemoryManager: 64102862 bytes of memory were used by task 4 but are not associated with specific consumers
23/06/14 19:38:57 INFO TaskMemoryManager: 5500051918 bytes of memory are used for execution and 38002226 bytes of memory are used for storage
23/06/14 19:38:57 WARN BlockManager: Block rdd_27_0 could not be removed as it was not found on disk or in memory

and executor is killed with

Diagnostics: [2023-05-28 16:24:44.080]Container killed on request. Exit code is 143\n[2023-05-28 16:24:44.080]Container exited with a non-zero exit code 143.

Below is the screenshot when job works fine.

Below is the screenshot when job fails

below is the screenshot for storage

code

Dataset<Row> currentDataSet = hdfsHandler.loadDataSet(sparkSession, "somelocation");
currentDataSet.createOrReplaceTempView(CURRENT_USER_VIEW);
currentDataSet.cache() //<- this causes the issue
Dataset<Row> previousDataSet = hdfsHandler.loadDataSet(sparkSession, "somelocation2);
previousDataSet.createOrReplaceTempView(PREVIOUS_USER_VIEW);
previousDataSet.cache() //<- this causes the issue

String currentRunColumn = "c.".concat("userid");
String previousRunColumn = "p.".concat("userid");

Dataset<Row> addedRecordDataSets = sparkSession.sql("SELECT " + currentRunColumn + " FROM " +
        CURRENT_USER_VIEW + " AS c " +
        " LEFT JOIN " + PREVIOUS_USER_VIEW + " AS p " +
        " ON " + currentRunColumn + " == " + previousRunColumn +
        " WHERE " + previousRunColumn + " IS NULL ");

dataSet.coalesce(1).persist(StorageLevel.DISK_ONLY()).foreachPartition(persist());

As par my understanding of above logs, at the time of exception execution memory is around 5gb and while it is trying to get more allocation it fails and the task fails. But can someone please help me understand how 41mb data can result into 5gb of execution memory? Is there a way to estimate the memory required for execution? I have gone through this and this amazing resources.

Update
Even though the data size is very small, peak execution memory is 48.1 GB. But the job does not die even when executor size is only 3gb. but when i add inmemory cache even though very small, the job dies with OOM (even when executor size is 10gb, hence execution size ~5gb), i am suspecting some bug wrt evicting cache? i am using 2.4.x of spark.

The root cause of the issue in our case was using .cache(), when i changed the .cache to .persist(Disk) or removed it all together, i did not face the issue.

Upon digging deeper this issues seems to be related to

https://issues.apache.org/jira/browse/SPARK-21492

So

a. updating to spark 2.4.5 or 3.0.0 could fix the issue.

b. remove .cache() on dataset.

UPDATE

i checked in spark 3.1.5, and no longer seeing the issue.