[SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..)

[hvanhovell]

What changes were proposed in this pull request?

This PR reduces the number of configuration RPCs needed for building a LocalRelation in the Scala client to 1.

Why are the changes needed?

This reduces the number of RPCs between the Scala client and the server when building a LocalRelation. This is more efficient, and better for performance.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added tests for RuntimeConfig.getMap(..).

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

No.

[heyihong]

nit: getMap is a bit vague. It indicates that the return type is a Map but doesn’t say much about its behavior. In my opinion, batchGet might be clearer, as it suggests that the method is used for batching config requests and improving efficiency.

[HyukjinKwon]

cc @zhengruifeng since you did the same thing in python side

[zhengruifeng]

it seems timeZoneId and largeVarTypes also need config RPCs

[LuciferYang]

From the definition, this is a longConf, so should we call toLong here?

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

Lines 6243 to 6254 in 14cc174

	val LOCAL_RELATION_CHUNK_SIZE_BYTES =
	buildConf(SqlApiConfHelper.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY)
	.doc("The chunk size in bytes when splitting ChunkedCachedLocalRelation.data " +
	"into batches. A new chunk is created when either " +
	"spark.sql.session.localRelationChunkSizeBytes " +
	"or spark.sql.session.localRelationChunkSizeRows is reached. " +
	"Limited by the spark.sql.session.localRelationBatchOfChunksSizeBytes, " +
	"a minimum of the two confs is used to determine the chunk size.")
	.version("4.1.0")
	.longConf
	.checkValue(_ > 0, "The chunk size in bytes must be positive")
	.createWithDefault(16 * 1024 * 1024L)

[LuciferYang]

We have a no-ansi daily test pipeline, and this case might fail in such a scenario:

SPARK_ANSI_SQL_MODE="false" build/sbt  clean "connect-client-jvm/testOnly org.apache.spark.sql.connect.ClientE2ETestSuite"

[info] - RuntimeConfig.get multiple keys *** FAILED *** (4 milliseconds)
[info]   Map("spark.sql.ansi.enabled" -> "false", "spark.sql.session.timeZone" -> "Asia/Shanghai", "spark.sql.binaryOutputStyle" -> "") did not equal Map("spark.sql.ansi.enabled" -> "true", "spark.sql.session.timeZone" -> "Asia/Shanghai", "spark.sql.binaryOutputStyle" -> "") (ClientE2ETestSuite.scala:1066)
[info]   Analysis:
[info]   Map("spark.sql.ansi.enabled": "false" -> "true")
[info]   org.scalatest.exceptions.TestFailedException:
[info]   at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
[info]   at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
[info]   at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
[info]   at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
[info]   at org.apache.spark.sql.connect.ClientE2ETestSuite.$anonfun$new$139(ClientE2ETestSuite.scala:1066)