This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-swift.git
The following commit(s) were added to refs/heads/main by this push:
new bc76bd1 [SPARK-58345] Handle Arrow message load failures in
`DataFrame.execute`
bc76bd1 is described below
commit bc76bd13c6f654888d92992955262f860d64d58b
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Fri Jul 24 19:56:28 2026 -0700
[SPARK-58345] Handle Arrow message load failures in `DataFrame.execute`
### What changes were proposed in this pull request?
This PR aims to fix `DataFrame.execute()` to throw an error when
`ArrowReader.fromMessage` fails, instead of ignoring the returned `Result`. In
addition, `ArrowReader.fromMessage` now returns `.failure` instead of
force-unwrapping `result.messageSchema!` when a record batch arrives before the
schema.
### Why are the changes needed?
Previously, a schema load failure (e.g., an unsupported Arrow type) crashed
the process on the force-unwrap, and a record-batch load failure silently
dropped data from `collect()` results.
### Does this PR introduce _any_ user-facing change?
No. Failures which previously crashed or silently dropped data now throw an
error.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #446 from dongjoon-hyun/SPARK-58345.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/ArrowReader.swift | 5 ++++-
Sources/SparkConnect/DataFrame+Actions.swift | 12 ++++++++++--
2 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/Sources/SparkConnect/ArrowReader.swift
b/Sources/SparkConnect/ArrowReader.swift
index c51569e..f3d3afd 100644
--- a/Sources/SparkConnect/ArrowReader.swift
+++ b/Sources/SparkConnect/ArrowReader.swift
@@ -478,9 +478,12 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
return .failure(error)
}
case .recordbatch:
+ guard let messageSchema = result.messageSchema, let arrowSchema =
result.schema else {
+ return .failure(.invalid("Record batch received before schema"))
+ }
let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self)!
let recordBatchResult = try loadRecordBatch(
- rbMessage, schema: result.messageSchema!, arrowSchema: result.schema!,
+ rbMessage, schema: messageSchema, arrowSchema: arrowSchema,
data: dataBody, messageEndOffset: 0
)
switch recordBatchResult {
diff --git a/Sources/SparkConnect/DataFrame+Actions.swift
b/Sources/SparkConnect/DataFrame+Actions.swift
index 5b5299a..455ed38 100644
--- a/Sources/SparkConnect/DataFrame+Actions.swift
+++ b/Sources/SparkConnect/DataFrame+Actions.swift
@@ -87,8 +87,16 @@ extension DataFrame {
// Read ArrowBatches
let reader = ArrowReader()
let arrowResult = ArrowReader.makeArrowReaderResult()
- _ = reader.fromMessage(schema, dataBody: Data(), result:
arrowResult)
- _ = reader.fromMessage(dataHeader, dataBody: dataBody, result:
arrowResult)
+ if case .failure(let error) = reader.fromMessage(
+ schema, dataBody: Data(), result: arrowResult)
+ {
+ throw error
+ }
+ if case .failure(let error) = reader.fromMessage(
+ dataHeader, dataBody: dataBody, result: arrowResult)
+ {
+ throw error
+ }
await self.addBatches(arrowResult.batches)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
