You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "hvanhovell (via GitHub)" <gi...@apache.org> on 2025/12/22 18:17:44 UTC
[PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
hvanhovell opened a new pull request, #53568:
URL: https://github.com/apache/spark/pull/53568
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "heyihong (via GitHub)" <gi...@apache.org>.
heyihong commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641365853
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/RuntimeConfig.scala:
##########
@@ -116,6 +116,18 @@ class RuntimeConfig private[sql] (client: SparkConnectClient)
}
}
+ private[connect] def getMap(keys: String*): Map[String, String] = {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641778760
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -117,11 +117,16 @@ class SparkSession private[sql] (
private def createDataset[T](encoder: AgnosticEncoder[T], data: Iterator[T]): Dataset[T] = {
newDataset(encoder) { builder =>
if (data.nonEmpty) {
- val threshold = conf.get(SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY).toInt
- val maxChunkSizeRows = conf.get(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_ROWS_KEY).toInt
- val maxChunkSizeBytes = conf.get(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY).toInt
+ val confs = conf.getMap(
+ SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY,
+ SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_ROWS_KEY,
+ SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY,
+ 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
Review Comment:
From the definition, this is a longConf, so should we call `toLong` here?
https://github.com/apache/spark/blob/14cc174a4e4cf3fef598587749ee95e400be62ab/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L6243-L6254
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641813111
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala:
##########
@@ -1053,6 +1053,19 @@ class ClientE2ETestSuite
assert(!spark.conf.contains("nope"))
}
+ test("RuntimeConfig.get multiple keys") {
+ assert(spark.conf.getMap().isEmpty)
+ val result = spark.conf.getMap(
+ "spark.sql.ansi.enabled",
Review 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)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641666485
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
Review Comment:
it seems `timeZoneId` and `largeVarTypes` also need config RPCs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641813111
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala:
##########
@@ -1053,6 +1053,19 @@ class ClientE2ETestSuite
assert(!spark.conf.contains("nope"))
}
+ test("RuntimeConfig.get multiple keys") {
+ assert(spark.conf.getMap().isEmpty)
+ val result = spark.conf.getMap(
+ "spark.sql.ansi.enabled",
Review Comment:
We have a pipeline with a `no-ansi` configuration, 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)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #53568: [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..)
URL: https://github.com/apache/spark/pull/53568
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54804][CONNECT] Reduce conf RPCs SparkSession.createDataset(..) [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #53568:
URL: https://github.com/apache/spark/pull/53568#discussion_r2641778760
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -117,11 +117,16 @@ class SparkSession private[sql] (
private def createDataset[T](encoder: AgnosticEncoder[T], data: Iterator[T]): Dataset[T] = {
newDataset(encoder) { builder =>
if (data.nonEmpty) {
- val threshold = conf.get(SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY).toInt
- val maxChunkSizeRows = conf.get(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_ROWS_KEY).toInt
- val maxChunkSizeBytes = conf.get(SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY).toInt
+ val confs = conf.getMap(
+ SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY,
+ SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_ROWS_KEY,
+ SqlApiConf.LOCAL_RELATION_CHUNK_SIZE_BYTES_KEY,
+ 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
Review Comment:
From the definition, this is a longConf.
https://github.com/apache/spark/blob/14cc174a4e4cf3fef598587749ee95e400be62ab/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L6243-L6254
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org