[SPARK-47804] Add Dataframe cache debug log

[anchovYu]

What changes were proposed in this pull request?

This PR adds a debug log for Dataframe cache that uses SQL conf to turn on. It logs necessary information on

• cache hit during cache application (the application happens basically on every query)
• cache miss
• adding new cache entries
• removing cache entries (including clear all entries)

Because every query applies cache, this log could be huge and should be only turned on during some debugging process, and should not enabled by default in production.

Example:

spark.conf.set("spark.sql.dataframeCache.logLevel", "warn")
val df = spark.range(1, 10)


df.collect()
{"ts":"2024-04-10T16:41:10.010-0700","level":"WARN","msg":"Dataframe cache miss for input plan:\nRange (1, 10, step=1, splits=Some(10))\n","logger":"org.apache.spark.sql.execution.CacheManager"}
{"ts":"2024-04-10T16:41:10.010-0700","level":"WARN","msg":"Last 20 Dataframe cache entry logical plans:\n[]","logger":"org.apache.spark.sql.execution.CacheManager"}

df.cache()
{"ts":"2024-04-10T16:42:18.647-0700","level":"WARN","msg":"Dataframe cache miss for input plan:\nRange (1, 10, step=1, splits=Some(10))\n","logger":"org.apache.spark.sql.execution.CacheManager"}
{"ts":"2024-04-10T16:42:18.647-0700","level":"WARN","msg":"Last 20 Dataframe cache entry logical plans:\n[]","logger":"org.apache.spark.sql.execution.CacheManager"}
{"ts":"2024-04-10T16:42:18.662-0700","level":"WARN","msg":"Added Dataframe cache entry:\nCachedData(\nlogicalPlan=Range (1, 10, step=1, splits=Some(10))\n\nInMemoryRelation=InMemoryRelation [id#2L], StorageLevel(disk, memory, deserialized, 1 replicas)\n   +- *(1) Range (1, 10, step=1, splits=10)\n)\n","logger":"org.apache.spark.sql.execution.CacheManager"}


df.count()
{"ts":"2024-04-10T16:43:36.033-0700","level":"WARN","msg":"Dataframe cache hit for input plan:\nRange (1, 10, step=1, splits=Some(10))\nmatched with cache entry:\nCachedData(\nlogicalPlan=Range (1, 10, step=1, splits=Some(10))\n\nInMemoryRelation=InMemoryRelation [id#2L], StorageLevel(disk, memory, deserialized, 1 replicas)\n   +- *(1) Range (1, 10, step=1, splits=10)\n)\n","logger":"org.apache.spark.sql.execution.CacheManager"}
{"ts":"2024-04-10T16:43:36.041-0700","level":"WARN","msg":"Dataframe cache hit plan change summary:\n Aggregate [count(1) AS count#13L]           Aggregate [count(1) AS count#13L]\n!+- Range (1, 10, step=1, splits=Some(10))   +- InMemoryRelation [id#2L], StorageLevel(disk, memory, deserialized, 1 replicas)\n!                                                  +- *(1) Range (1, 10, step=1, splits=10)","logger":"org.apache.spark.sql.execution.CacheManager"}


df.unpersist()
{"ts":"2024-04-10T16:44:15.965-0700","level":"WARN","msg":"Removed 1 Dataframe cache entries, with logical plans being \n[Range (1, 10, step=1, splits=Some(10))\n]","logger":"org.apache.spark.sql.execution.CacheManager"}

Why are the changes needed?

Easier debugging.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Run local spark shell.

Was this patch authored or co-authored using generative AI tooling?

No.

[gengliangwang]

I think we need to migrate to the structured logging API here.
Framework: #45729
Example migration PR: #45834

[gengliangwang]

Will we need to debug cache table as well? Shall we rename the config as
spark.sql.cache.logLevel

[gengliangwang]

Also, let's make it an internal conf since it is for developers.

[anchovYu]

I kept the Dataframe cache naming to differentiate it from the RDD cache.

[gengliangwang]

RDD is a Spark core concept. Anyway I respect your choice here.

[gengliangwang]

Thanks, merging to master