[SPARK-33206][CORE][3.2] Fix shuffle index cache weight calculation for small index files

[attilapiros]

What changes were proposed in this pull request?

Increasing the shuffle index weight with a constant number to avoid underestimating retained memory size caused by the bookkeeping objects: the java.io.File (depending on the path ~ 960 bytes) object and the ShuffleIndexInformation object (~180 bytes).

Why are the changes needed?

Underestimating cache entry size easily can cause OOM in the Yarn NodeManager.
In the following analyses of a prod issue (HPROF file) we can see the leak suspect Guava's LocalCache$Segment objects:

Going further we can see a ShuffleIndexInformation for a small index file (16 bytes) but the retained heap memory is 1192 bytes:

Finally we can see this is very common within this heap dump (using MAT's Object Query Language):

I have even exported the data to a CSV and done some calculations with awk:

$ tail -n+2 export.csv | awk -F, 'BEGIN { numUnderEstimated=0; } { sumOldSize += $1; corrected=$1 + 1176; sumCorrectedSize += corrected; sumRetainedMem += $2; if (corrected < $2) numUnderEstimated+=1; } END { print "sum old size: " sumOldSize / 1024 / 1024   " MB, sum corrected size: " sumCorrectedSize / 1024 / 1024 " MB, sum retained memory:" sumRetainedMem / 1024 / 1024  " MB, num under estimated: " numUnderEstimated }'

It gives the followings:

sum old size: 76.8785 MB, sum corrected size: 1066.93 MB, sum retained memory:1064.47 MB, num under estimated: 0

So using the old calculation we were at 7.6.8 MB way under the default cache limit (100 MB).
Using the correction (applying 1176 as increment to the size) we are at 1066.93 MB (~1GB) which is close to the real retained sum heap: 1064.47 MB (~1GB) and there is no entry which was underestimated.

But we can go further and get rid of java.io.File completely and store the ShuffleIndexInformation for the file path.
This way not only the cache size estimate is improved but the its size is decreased as well.
Here the path size is not counted into the cache size as that string is interned.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

With the calculations above.

[attilapiros]

Backport of #35559 to 3.2

[dongjoon-hyun]

Thank you for making a backport, @attilapiros .

[attilapiros]

Thank you for making a backport, @attilapiros .

@dongjoon-hyun Tomorrow I will open one against branch-3.1.

[dongjoon-hyun]

Thanks!

cc @mridulm , @srowen , @Ngone51 , @wypoon too

[dongjoon-hyun]

+1, LGTM. The linter job error is irrelevant to this PR.

pip._vendor.urllib3.exceptions.ProtocolError: ("Connection broken: ConnectionResetError(104, 'Connection reset by peer')", ConnectionResetError(104, 'Connection reset by peer'))
[95](https://github.com/attilapiros/spark/runs/5395854606?check_suite_focus=true#step:8:95)
WARNING: You are using pip version 21.1.2; however, version 22.0.3 is available.

If you are ready, could you switch into a normal PR from Draft, @attilapiros ?

[dongjoon-hyun]

Thank you, @attilapiros and @srowen .
Merged to branch-3.2 for Apache Spark 3.2.2.

[mridulm]

Late LGTM, thanks @attilapiros !