Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-49673][CONNECT] Increase CONNECT_GRPC_ARROW_MAX_BATCH_SIZE to 0.7 * CONNECT_GRPC_MAX_MESSAGE_SIZE #48122

Closed
wants to merge 8 commits into from
Closed

[SPARK-49673][CONNECT] Increase CONNECT_GRPC_ARROW_MAX_BATCH_SIZE to 0.7 * CONNECT_GRPC_MAX_MESSAGE_SIZE #48122

Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a68f280
increase maxBatchSize to 0.7 * grpc message size limit for sqlCommand…
dillitz Sep 16, 2024
18a330f
remove unused import
dillitz Sep 16, 2024
c04795c
Revert "remove unused import"
dillitz Sep 17, 2024
cf006e8
Revert "increase maxBatchSize to 0.7 * grpc message size limit for sq…
dillitz Sep 17, 2024
01eddd3
change CONNECT_GRPC_ARROW_MAX_BATCH_SIZE to CONNECT_GRPC_MAX_MESSAGE_…
dillitz Sep 17, 2024
4eb6711
add E2E test
dillitz Sep 17, 2024
ca10ef9
adjust client's maxMessageSize
dillitz Sep 17, 2024
b0bfb27
linter
dillitz Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions ...nect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import io.grpc.{Context, Status, StatusRuntimeException}
import io.grpc.stub.StreamObserver
import org.apache.commons.lang3.exception.ExceptionUtils

import org.apache.spark.{Partition, SparkEnv, TaskContext}
import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.annotation.{DeveloperApi, Since}
import org.apache.spark.api.python.{PythonEvalType, SimplePythonFunction}
import org.apache.spark.connect.proto
Expand Down Expand Up @@ -61,7 +61,7 @@ import org.apache.spark.sql.catalyst.trees.PySparkCurrentOrigin
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
import org.apache.spark.sql.connect.common.{DataTypeProtoConverter, ForeachWriterPacket, InvalidPlanInput, LiteralValueProtoConverter, StorageLevelProtoConverter, StreamingListenerPacket, UdfPacket}
import org.apache.spark.sql.connect.config.Connect.CONNECT_GRPC_ARROW_MAX_BATCH_SIZE
import org.apache.spark.sql.connect.common.config.ConnectCommon
import org.apache.spark.sql.connect.plugin.SparkConnectPluginRegistry
import org.apache.spark.sql.connect.service.{ExecuteHolder, SessionHolder, SparkConnectService}
import org.apache.spark.sql.connect.utils.MetricGenerator
Expand Down Expand Up @@ -2498,7 +2498,7 @@ class SparkConnectPlanner(
if (isCommand) {
// Convert the results to Arrow.
val schema = df.schema
val maxBatchSize = (SparkEnv.get.conf.get(CONNECT_GRPC_ARROW_MAX_BATCH_SIZE) * 0.7).toLong
val maxBatchSize = (ConnectCommon.CONNECT_GRPC_MAX_MESSAGE_SIZE * 0.7).toLong
dillitz marked this conversation as resolved.
Show resolved Hide resolved
val timeZoneId = session.sessionState.conf.sessionLocalTimeZone

// Convert the data.
Expand Down