[SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) #53568
Conversation
| } | ||
| } | ||
|
|
||
| private[connect] def getMap(keys: String*): Map[String, String] = { |
There was a problem hiding this comment.
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
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
cc @zhengruifeng since you did the same thing in python side
| maxRecordsPerBatch = maxChunkSizeRows, | ||
| maxBatchSize = math.min(maxChunkSizeBytes, maxBatchOfChunksSize), | ||
| timeZoneId = timeZoneId, | ||
| largeVarTypes = largeVarTypes, |
There was a problem hiding this comment.
it seems timeZoneId and largeVarTypes also need config RPCs
| SqlApiConf.LOCAL_RELATION_BATCH_OF_CHUNKS_SIZE_BYTES_KEY) | ||
| val threshold = confs(SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY).toInt | ||
| val maxChunkSizeRows = confs(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_ROWS_KEY).toInt | ||
| val maxChunkSizeBytes = confs(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY).toInt |
There was a problem hiding this comment.
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
| test("RuntimeConfig.get multiple keys") { | ||
| assert(spark.conf.getMap().isEmpty) | ||
| val result = spark.conf.getMap( | ||
| "spark.sql.ansi.enabled", |
There was a problem hiding this comment.
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)
5 participants
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.