[SPARK-50324][PYTHON][CONNECT] Make createDataFrame trigger Config RPC at most once

[zhengruifeng]

What changes were proposed in this pull request?

Get all configs in batch

Why are the changes needed?

there are too many related configs in createDataFrame, they are fetched one by one (or group by group) in different branches:
1, it is possible no Config RPC is triggered, e.g. in this branch:

spark/python/pyspark/sql/connect/session.py

Lines 502 to 509 in 2633035

	elif isinstance(data, Sized) and len(data) == 0:
	if _schema is not None:
	return DataFrame(LocalRelation(table=None, schema=_schema.json()), self)
	else:
	raise PySparkValueError(
	errorClass="CANNOT_INFER_EMPTY_SCHEMA",
	messageParameters={},
	)

2, multiple Config RPCs for different configs, e.g. in this branch:

spark/python/pyspark/sql/connect/session.py

Lines 599 to 601 in 2633035

	prefer_timestamp_ntz = is_timestamp_ntz_preferred()
	(timezone,) = self._client.get_configs("spark.sql.session.timeZone")

Does this PR introduce any user-facing change?

no

How was this patch tested?

ci

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

no

[xinrong-meng]

LGTM thank you!

[HyukjinKwon]

Shall we just get all the confs in batch eagerly? Seems like we should get the conf once anyway.

[zhengruifeng]

there are 2 cases that we don't need the configs:
1, the local data is empty, and the schema is specified, it returns a valid empty df;
2, the creation fails due to some assertions

[HyukjinKwon]

I think another way of doing this is to maintain a sized dictionary in Python side, and cache the value retrieved within the Python side.

e.g.,

• spark.get("a")

  • look up cacehd["a"] = v
    • if not, spark.get("a")

• spark.set("a", aa")

  • empty cache cached["a"]
  • spark.set("a")

and create a dictioanry with TTL and max size

[HyukjinKwon]

which will work all for Spark Classic and Spark Connect.

[zhengruifeng]

I think another way of doing this is to maintain a sized dictionary in Python side, and cache the value retrieved within the Python side.

e.g.,

• spark.get("a")

  • look up cacehd["a"] = v

    • if not, spark.get("a")

• spark.set("a", aa")

  • empty cache cached["a"]
  • spark.set("a")

and create a dictioanry with TTL and max size

A problem is that spark.conf is not the only entry point for config operations.
Users can also set/unset config by spark.sql

[zhengruifeng]

It seems we don't need this helper class to achieve the goal, will have another try

[zhengruifeng]

thanks, merged to master