You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "dongjoon-hyun (via GitHub)" <gi...@apache.org> on 2026/10/16 18:09:20 UTC
[PR] [SPARK-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
dongjoon-hyun opened a new pull request, #597:
URL: https://github.com/apache/spark-connect-swift/pull/597
### What changes were proposed in this pull request?
This PR aims to align the gRPC message size limit and the Arrow result handling with the Scala and Python Spark Connect clients.
1. Set the default `grpc_max_message_size` to 128 MiB (`SparkConnectClient.DEFAULT_GRPC_MAX_MESSAGE_SIZE`). `SparkConnectClient.maxMessageSize` is no longer optional, and the `ServiceConfig` always applies the limit to both request and response messages.
2. Send `ResultChunkingOptions(allow_arrow_batch_chunking = true)` in every `ExecutePlanRequest` so that the server can split large Arrow batches into chunks.
3. Assemble the chunks in `DataFrame.execute()` using `chunk_index` and `num_chunks_in_batch` before parsing the Arrow IPC stream. An out-of-order chunk fails with `InvalidArrowData`.
4. Count `row_count` only for the first chunk (`chunk_index == 0`) in `executeAndCount()`, because the server sets the same `row_count` on all chunks of a batch.
5. Add the `ResultChunkingOptions` type alias.
`preferred_arrow_chunk_size` is not exposed in this PR.
### Why are the changes needed?
Without `grpc_max_message_size`, the client is limited to the gRPC default of 4 MiB, so a response larger than 4 MiB fails with `Received message larger than max`. The Scala client and the Python client both default to 128 MiB. Python has used 128 MiB since [SPARK-42816](https://github.com/apache/spark/commit/da19e0de05b) (2023-03-21, v3.5.0).
Even with a larger limit, a single Arrow batch can exceed it. Result chunking, enabled by default in the Scala and Python clients, lets the server split such batches.
### Does this PR introduce _any_ user-facing change?
Yes.
- When `grpc_max_message_size` is not specified, the limit changes from 4 MiB to 128 MiB. Responses larger than 4 MiB now succeed instead of failing.
- Large Arrow batches are now received in chunks and transparently assembled.
### How was this patch tested?
- Updated `parameterWithoutGrpcMaxMessageSize` to expect the 128 MiB default in `maxMessageSize` and in the `ServiceConfig`.
- Added `resultChunking` to `SparkConnectClientTests` to verify the request option.
- Confirmed that `swift build` and the test target compile.
- The integration tests were not run locally because no Spark Connect server was available. They are expected to run in CI.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#discussion_r4298485569
##########
Sources/SparkConnect/DataFrame+Actions.swift:
##########
@@ -38,15 +38,30 @@ extension DataFrame {
/// Batches are local to each call since concurrent actions can interleave on this actor.
private func execute() async throws -> [RecordBatch] {
let batches = Mutex([RecordBatch]())
+ // The chunks of the Arrow batch being received. The server splits a large batch into chunks.
+ let chunks = Mutex([Data]())
try await spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
m in
if m.hasSchema {
// The original schema should arrive before ArrowBatches
try await self.setSchema(DataType(m.schema))
}
- let ipcStreamBytes = m.arrowBatch.data
- if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
+ if !m.arrowBatch.data.isEmpty && m.arrowBatch.rowCount > 0,
+ let ipcStreamBytes = try chunks.withLock({ chunks -> Data? in
+ guard m.arrowBatch.chunkIndex == chunks.count else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ chunks.append(m.arrowBatch.data)
+ // `numChunksInBatch` is missing when the batch is not chunked.
+ guard m.arrowBatch.numChunksInBatch == 0 || chunks.count == m.arrowBatch.numChunksInBatch
+ else {
+ return nil
+ }
+ defer { chunks.removeAll() }
+ return chunks.count == 1 ? chunks[0] : chunks.reduce(into: Data()) { $0.append($1) }
Review Comment:
No problem, thanks for catching that. Switched to `reduce(into:)` with the capacity reserved up front in 90da317.
##########
Tests/SparkConnectTests/DataFrameTests.swift:
##########
@@ -704,6 +704,22 @@ struct DataFrameTests {
await spark.stop()
}
+ @Test
+ func resultChunking() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await isSparkVersionAtLeast(spark.version, "4.1") {
+ let df = try await spark.range(10000).selectExpr("id", "repeat('a', 100) AS s")
+ let expected = try await df.collect()
+ // Force the server to split each Arrow batch into chunks.
+ try await spark.conf.set("spark.connect.session.resultChunking.maxChunkSize", "1024")
+ #expect(try await df.collect() == expected)
+ #expect(try await df.count() == 10000)
Review Comment:
Good point. Replaced the two checks with `executeAndCount() == 10000` and switched `DataFrameTests` to `@testable import SparkConnect` in 90da317.
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#discussion_r4298338982
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -109,17 +112,13 @@ public actor SparkConnectClient {
self.transportSecurity = .plaintext
}
self.maxMessageSize = maxMessageSize
- if let maxMessageSize {
- // An empty service name is the default configuration for all services and methods.
- self.serviceConfig = ServiceConfig(methodConfig: [
- MethodConfig(
- names: [MethodConfig.Name(service: "")],
- maxRequestMessageBytes: maxMessageSize,
- maxResponseMessageBytes: maxMessageSize)
- ])
- } else {
- self.serviceConfig = ServiceConfig()
- }
+ // An empty service name is the default configuration for all services and methods.
Review Comment:
Good catch, thanks. Added the note in fefb833.
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -486,6 +485,12 @@ public actor SparkConnectClient {
request.operationID = UUID().uuidString
request.tags = Array(tags)
request.plan = plan
+ // Allow the server to split large Arrow batches into chunks which `DataFrame` assembles.
+ var chunkingOptions = ResultChunkingOptions()
+ chunkingOptions.allowArrowBatchChunking = true
Review Comment:
Thanks for the suggestion. I'd like to keep this PR aligned with Scala and Python, which don't handle this either. I'll do it in a follow-up PR.
##########
Sources/SparkConnect/DataFrame+Actions.swift:
##########
@@ -38,15 +38,30 @@ extension DataFrame {
/// Batches are local to each call since concurrent actions can interleave on this actor.
private func execute() async throws -> [RecordBatch] {
let batches = Mutex([RecordBatch]())
+ // The chunks of the Arrow batch being received. The server splits a large batch into chunks.
+ let chunks = Mutex([Data]())
try await spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
m in
if m.hasSchema {
// The original schema should arrive before ArrowBatches
try await self.setSchema(DataType(m.schema))
}
- let ipcStreamBytes = m.arrowBatch.data
- if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
+ if !m.arrowBatch.data.isEmpty && m.arrowBatch.rowCount > 0,
+ let ipcStreamBytes = try chunks.withLock({ chunks -> Data? in
+ guard m.arrowBatch.chunkIndex == chunks.count else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ chunks.append(m.arrowBatch.data)
+ // `numChunksInBatch` is missing when the batch is not chunked.
+ guard m.arrowBatch.numChunksInBatch == 0 || chunks.count == m.arrowBatch.numChunksInBatch
+ else {
+ return nil
+ }
+ defer { chunks.removeAll() }
+ return chunks.count == 1 ? chunks[0] : chunks.reduce(into: Data()) { $0.append($1) }
Review Comment:
Done in fefb833.
##########
Tests/SparkConnectTests/SparkConnectClientTests.swift:
##########
@@ -123,6 +125,16 @@ struct SparkConnectClientTests {
await client.stop()
}
+ @Test
+ func resultChunking() async throws {
+ let client = try SparkConnectClient(remote: TEST_REMOTE)
+ try await client.connect(UUID().uuidString)
+ let request = await client.getExecutePlanRequest(await client.getPlanRange(0, 1, 1))
+ #expect(request.requestOptions.count == 1)
Review Comment:
Done in fefb833.
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -28,14 +28,17 @@ import Synchronization
/// Conceptually the remote spark session that communicates with the server
public actor SparkConnectClient {
+ /// The default `grpc_max_message_size`, 128 MiB, which is the same with Scala and Python clients.
Review Comment:
Done in fefb833.
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#discussion_r4298463584
##########
Sources/SparkConnect/DataFrame+Actions.swift:
##########
@@ -38,15 +38,30 @@ extension DataFrame {
/// Batches are local to each call since concurrent actions can interleave on this actor.
private func execute() async throws -> [RecordBatch] {
let batches = Mutex([RecordBatch]())
+ // The chunks of the Arrow batch being received. The server splits a large batch into chunks.
+ let chunks = Mutex([Data]())
try await spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
m in
if m.hasSchema {
// The original schema should arrive before ArrowBatches
try await self.setSchema(DataType(m.schema))
}
- let ipcStreamBytes = m.arrowBatch.data
- if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
+ if !m.arrowBatch.data.isEmpty && m.arrowBatch.rowCount > 0,
+ let ipcStreamBytes = try chunks.withLock({ chunks -> Data? in
+ guard m.arrowBatch.chunkIndex == chunks.count else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ chunks.append(m.arrowBatch.data)
+ // `numChunksInBatch` is missing when the batch is not chunked.
+ guard m.arrowBatch.numChunksInBatch == 0 || chunks.count == m.arrowBatch.numChunksInBatch
+ else {
+ return nil
+ }
+ defer { chunks.removeAll() }
+ return chunks.count == 1 ? chunks[0] : chunks.reduce(into: Data()) { $0.append($1) }
Review Comment:
Sorry, my earlier suggestion was wrong. `FlattenSequence<[Data]>` doesn't provide contiguous storage, and its `underestimatedCount` is 0. So `Data.init(_: some Sequence<UInt8>)` falls back to copying byte by byte (the `slowElements` path in swift-foundation). This branch only runs for batches over about 115 MiB, so that path matters. The original `reduce(into:)` with `append` copied whole chunks. Reserving capacity up front keeps that and avoids regrowth:
```swift
return chunks.count == 1
? chunks[0]
: chunks.reduce(into: Data(capacity: chunks.reduce(0) { $0 + $1.count })) { $0.append($1) }
```
Reverting to the original `reduce(into: Data())` is also fine.
##########
Tests/SparkConnectTests/DataFrameTests.swift:
##########
@@ -704,6 +704,22 @@ struct DataFrameTests {
await spark.stop()
}
+ @Test
+ func resultChunking() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await isSparkVersionAtLeast(spark.version, "4.1") {
+ let df = try await spark.range(10000).selectExpr("id", "repeat('a', 100) AS s")
+ let expected = try await df.collect()
+ // Force the server to split each Arrow batch into chunks.
+ try await spark.conf.set("spark.connect.session.resultChunking.maxChunkSize", "1024")
+ #expect(try await df.collect() == expected)
+ #expect(try await df.count() == 10000)
Review Comment:
nit: These two checks don't exercise the `chunkIndex == 0` counting in `executeAndCount()`. `count()` is computed on the server and returns a single row, and `isEmpty()` uses `limit(1)`, where over-counting can't turn a non-zero result into zero. To cover it, `#expect(try await df.executeAndCount() == 10000)` would need `@testable import SparkConnect`. Otherwise it's fine to drop these two lines, since `collect()` already covers chunk assembly.
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #597: [SPARK-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking
URL: https://github.com/apache/spark-connect-swift/pull/597
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#issuecomment-6223195454
Thank you so much for the suggestion and approval, @viirya !
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#discussion_r4298259156
##########
Tests/SparkConnectTests/SparkConnectClientTests.swift:
##########
@@ -123,6 +125,16 @@ struct SparkConnectClientTests {
await client.stop()
}
+ @Test
+ func resultChunking() async throws {
+ let client = try SparkConnectClient(remote: TEST_REMOTE)
+ try await client.connect(UUID().uuidString)
+ let request = await client.getExecutePlanRequest(await client.getPlanRange(0, 1, 1))
+ #expect(request.requestOptions.count == 1)
Review Comment:
This will break as soon as another request option is added. How about checking for the option itself?
```swift
#expect(request.requestOptions.contains { $0.resultChunkingOptions.allowArrowBatchChunking })
```
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -28,14 +28,17 @@ import Synchronization
/// Conceptually the remote spark session that communicates with the server
public actor SparkConnectClient {
+ /// The default `grpc_max_message_size`, 128 MiB, which is the same with Scala and Python clients.
Review Comment:
nit: "the same with" → "the same as".
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -486,6 +485,12 @@ public actor SparkConnectClient {
request.operationID = UUID().uuidString
request.tags = Array(tags)
request.plan = plan
+ // Allow the server to split large Arrow batches into chunks which `DataFrame` assembles.
+ var chunkingOptions = ResultChunkingOptions()
+ chunkingOptions.allowArrowBatchChunking = true
Review Comment:
A question about smaller limits. If a user sets `grpc_max_message_size` below about 115 MiB (for example 64 MiB), the server still uses its default ~115 MiB chunks, so chunking won't help for batches between the two sizes. Scala and Python don't handle this either, so it's fine to leave as is. Would it be worth sending `preferred_arrow_chunk_size` (for example `maxMessageSize * 9 / 10`) when `maxMessageSize` is below the default? A follow-up would be fine too.
##########
Sources/SparkConnect/SparkConnectClient.swift:
##########
@@ -109,17 +112,13 @@ public actor SparkConnectClient {
self.transportSecurity = .plaintext
}
self.maxMessageSize = maxMessageSize
- if let maxMessageSize {
- // An empty service name is the default configuration for all services and methods.
- self.serviceConfig = ServiceConfig(methodConfig: [
- MethodConfig(
- names: [MethodConfig.Name(service: "")],
- maxRequestMessageBytes: maxMessageSize,
- maxResponseMessageBytes: maxMessageSize)
- ])
- } else {
- self.serviceConfig = ServiceConfig()
- }
+ // An empty service name is the default configuration for all services and methods.
Review Comment:
It's worth noting here why both limits must be set. In grpc-swift-nio-transport 2.11.0, the client's inbound deframer is sized from `maxRequestMessageBytes ?? 4 MiB` (`Connection.swift`, `maxPayloadSize: maxRequestSize`). `maxResponseMessageBytes` isn't used by the transport for responses. So the 128 MiB response limit takes effect only because the request limit is set too. If someone later drops the request limit, responses would silently fall back to 4 MiB. Something like:
```swift
// Note that grpc-swift-nio-transport limits inbound response messages by
// `maxRequestMessageBytes`, so both limits are required.
```
##########
Sources/SparkConnect/DataFrame+Actions.swift:
##########
@@ -38,15 +38,30 @@ extension DataFrame {
/// Batches are local to each call since concurrent actions can interleave on this actor.
private func execute() async throws -> [RecordBatch] {
let batches = Mutex([RecordBatch]())
+ // The chunks of the Arrow batch being received. The server splits a large batch into chunks.
+ let chunks = Mutex([Data]())
try await spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
m in
if m.hasSchema {
// The original schema should arrive before ArrowBatches
try await self.setSchema(DataType(m.schema))
}
- let ipcStreamBytes = m.arrowBatch.data
- if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
+ if !m.arrowBatch.data.isEmpty && m.arrowBatch.rowCount > 0,
+ let ipcStreamBytes = try chunks.withLock({ chunks -> Data? in
+ guard m.arrowBatch.chunkIndex == chunks.count else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ chunks.append(m.arrowBatch.data)
+ // `numChunksInBatch` is missing when the batch is not chunked.
+ guard m.arrowBatch.numChunksInBatch == 0 || chunks.count == m.arrowBatch.numChunksInBatch
+ else {
+ return nil
+ }
+ defer { chunks.removeAll() }
+ return chunks.count == 1 ? chunks[0] : chunks.reduce(into: Data()) { $0.append($1) }
Review Comment:
nit: For the multi-chunk branch, `Data(chunks.joined())` is shorter and avoids growing the buffer repeatedly:
```swift
return chunks.count == 1 ? chunks[0] : Data(chunks.joined())
```
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#issuecomment-6222779818
Thank you for the thorough review, @viirya. I addressed the comments in fefb833:
- Added `DataFrameTests.resultChunking` (gated on Spark 4.1+). It sets `spark.connect.session.resultChunking.maxChunkSize` to 1024 to force chunking and checks `collect()`, `count()` and `isEmpty()`.
- `execute()` now throws `InvalidArrowData` if the stream ends with an incomplete chunked batch.
- Added the note about `maxRequestMessageBytes` and `maxResponseMessageBytes`, simplified the chunk concatenation, relaxed the test assertion and fixed the typo.
`preferred_arrow_chunk_size` for a smaller `grpc_max_message_size` will be handled in a follow-up PR.
--
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-60298] Set default `grpc_max_message_size` to 128 MiB and support Arrow batch result chunking [spark-connect-swift]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #597:
URL: https://github.com/apache/spark-connect-swift/pull/597#issuecomment-6223212807
Merged to main
--
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