[SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests

[xi-db]

What changes were proposed in this pull request?

While building the DataFrame step by step, each time a new DataFrame is generated with an empty schema, which is lazily computed on access. However, if a user's code frequently accesses the schema of these new DataFrames using methods such as df.columns, it will result in a large number of Analyze requests to the server. Each time, the entire plan needs to be reanalyzed, leading to poor performance, especially when constructing highly complex plans.

Now, by introducing plan cache in SparkConnectPlanner, we aim to reduce the overhead of repeated analysis during this process. This is achieved by saving significant computation if the resolved logical plan of a subtree of can be cached.

A minimal example of the problem:

import pyspark.sql.functions as F
df = spark.range(10)
for i in range(200):
  if str(i) not in df.columns: # <-- The df.columns call causes a new Analyze request in every iteration
    df = df.withColumn(str(i), F.col("id") + i)
df.show() 

With this patch, the performance of the above code improved from ~110s to ~5s.

Why are the changes needed?

The performance improvement is huge in the above cases.

Does this PR introduce any user-facing change?

Yes, a static conf spark.connect.session.planCache.maxSize and a dynamic conf spark.connect.session.planCache.enabled are added.

• spark.connect.session.planCache.maxSize: Sets the maximum number of cached resolved logical plans in Spark Connect Session. If set to a value less or equal than zero will disable the plan cache
• spark.connect.session.planCache.enabled: When true, the cache of resolved logical plans is enabled if spark.connect.session.planCache.maxSize is greater than zero. When false, the cache is disabled even if spark.connect.session.planCache.maxSize is greater than zero. The caching is best-effort and not guaranteed.

How was this patch tested?

Some new tests are added in SparkConnectSessionHolderSuite.scala.

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

No.

[beliefer]

I submitted #43473 half year ago.
It seems not received enough attention.

[zhengruifeng]

cc @HyukjinKwon and @ueshin

[zhengruifeng]

I think we may want to exclude some kinds of plans:
1, LOCAL_RELATION, its size might be large;
2, CACHED_REMOTE_RELATION, it is already cached in dataFrameCache in the SessionHolder;
3, CATALOG, it may be a command like CREATE_TABLE/DROP_TEMP_VIEW , should not be skipped.

[vicennial]

@zhengruifeng Does point 3 extend to all DDL/DML commands?

[vicennial]

1, LOCAL_RELATION, its size might be large;
2, CACHED_REMOTE_RELATION, it is already cached in dataFrameCache in the SessionHolder;

We use a low default value for CONNECT_SESSION_PLAN_CACHE_SIZE, 5 entries to be specific, to prevent the cache from becoming too large. We haven't been too surgical in what kind of plans we cache to keep it as a simple wrapper in aiding common scenarios that may generate some repeated analysis/execute calls.

[xi-db]

3, CATALOG, it may be a command like CREATE_TABLE/DROP_TEMP_VIEW , should not be skipped.

Commands should be fine. Each command will get a new plan_id, so a new command won't hit the cache. Besides, we only cache plans with plan_id set.

[zhengruifeng]

yea, I forgot the new plan_id. Then it's fine

[HyukjinKwon]

Merged to master.