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 0a6f81a [SPARK-58051] Replace release-stripped asserts with runtime
validation
0a6f81a is described below
commit 0a6f81a55766db05dacf710b6637f35b7eda4b98
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Jul 8 13:58:27 2026 -0700
[SPARK-58051] Replace release-stripped asserts with runtime validation
### What changes were proposed in this pull request?
Replace release-stripped `assert(...)` calls with `guard`/`throw`
validation and add `SparkConnectError.InvalidArrowData`:
- `DataFrame.execute()` — validate Arrow IPC continuation tokens and
bounds-check server-supplied sizes before slicing.
- `DataFrame.show()` — require one row / one column before accessing
`rows[0]`.
- `DataFrameWriter.executeWriteOperation` — reject `path` and `tableName`
both set.
- `SparkSession.Builder.create()` — require `setConf` success.
### Why are the changes needed?
`assert(...)` is compiled out in release builds, so these checks never ran
in production. A malformed server response could trap while slicing on
unvalidated sizes. `guard`/`throw` makes the checks effective in release and
turns traps into recoverable errors.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #435 from dongjoon-hyun/SPARK-58051.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/DataFrame+Actions.swift | 31 +++++++++++++++++++++-------
Sources/SparkConnect/DataFrameWriter.swift | 4 +++-
Sources/SparkConnect/SparkConnectError.swift | 1 +
Sources/SparkConnect/SparkSession.swift | 4 +++-
4 files changed, 31 insertions(+), 9 deletions(-)
diff --git a/Sources/SparkConnect/DataFrame+Actions.swift
b/Sources/SparkConnect/DataFrame+Actions.swift
index 7962bcc..5b5299a 100644
--- a/Sources/SparkConnect/DataFrame+Actions.swift
+++ b/Sources/SparkConnect/DataFrame+Actions.swift
@@ -51,21 +51,37 @@ extension DataFrame {
let ipcStreamBytes = m.arrowBatch.data
if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
let IPC_CONTINUATION_TOKEN = Int32(-1)
+ let totalSize = Int64(ipcStreamBytes.count)
// Schema
- assert(ipcStreamBytes[0..<4].int32 == IPC_CONTINUATION_TOKEN)
+ guard totalSize >= 8,
+ ipcStreamBytes[0..<4].int32 == IPC_CONTINUATION_TOKEN
+ else {
+ throw SparkConnectError.InvalidArrowData
+ }
let schemaSize = Int64(ipcStreamBytes[4..<8].int32)
+ guard schemaSize >= 0, 8 + schemaSize + 4 <= totalSize else {
+ throw SparkConnectError.InvalidArrowData
+ }
let schema = Data(ipcStreamBytes[8..<(8 + schemaSize)])
// Arrow IPC Data
- assert(
- ipcStreamBytes[(8 + schemaSize)..<(8 + schemaSize + 4)].int32
- == IPC_CONTINUATION_TOKEN)
+ guard ipcStreamBytes[(8 + schemaSize)..<(8 + schemaSize + 4)].int32
+ == IPC_CONTINUATION_TOKEN
+ else {
+ throw SparkConnectError.InvalidArrowData
+ }
var pos: Int64 = 8 + schemaSize + 4
+ guard pos + 4 <= totalSize else {
+ throw SparkConnectError.InvalidArrowData
+ }
let dataHeaderSize = Int64(ipcStreamBytes[pos..<(pos + 4)].int32)
pos += 4
+ guard dataHeaderSize >= 0, pos + dataHeaderSize + 8 <= totalSize
else {
+ throw SparkConnectError.InvalidArrowData
+ }
let dataHeader = Data(ipcStreamBytes[pos..<(pos + dataHeaderSize)])
pos += dataHeaderSize
- let dataBodySize = Int64(ipcStreamBytes.count) - pos - 8
+ let dataBodySize = totalSize - pos - 8
let dataBody = Data(ipcStreamBytes[pos..<(pos + dataBodySize)])
// Read ArrowBatches
@@ -196,8 +212,9 @@ extension DataFrame {
/// - vertical: If set to true, prints output rows vertically (one line
per column value).
public func show(_ numRows: Int32, _ truncate: Int32, _ vertical: Bool =
false) async throws {
let rows = try await showString(numRows, truncate, vertical).collect()
- assert(rows.count == 1)
- assert(rows[0].length == 1)
+ guard rows.count == 1, rows[0].length == 1 else {
+ throw SparkConnectError.InvalidArrowData
+ }
print(try rows[0].get(0) as! String)
}
diff --git a/Sources/SparkConnect/DataFrameWriter.swift
b/Sources/SparkConnect/DataFrameWriter.swift
index df35c45..07a3f82 100644
--- a/Sources/SparkConnect/DataFrameWriter.swift
+++ b/Sources/SparkConnect/DataFrameWriter.swift
@@ -147,7 +147,9 @@ public actor DataFrameWriter: Sendable {
var write = f()
// Cannot both be set
- assert(!(!write.path.isEmpty && !write.table.tableName.isEmpty))
+ guard write.path.isEmpty || write.table.tableName.isEmpty else {
+ throw SparkConnectError.InvalidArgument
+ }
let plan = await self.df.getPlan() as! Plan
write.input = plan.root
diff --git a/Sources/SparkConnect/SparkConnectError.swift
b/Sources/SparkConnect/SparkConnectError.swift
index a0d4efe..45de0a9 100644
--- a/Sources/SparkConnect/SparkConnectError.swift
+++ b/Sources/SparkConnect/SparkConnectError.swift
@@ -23,6 +23,7 @@ public enum SparkConnectError: Error {
case ColumnNotFound
case DataSourceNotFound
case InvalidArgument
+ case InvalidArrowData
case InvalidSessionID
case InvalidType
case InvalidViewName
diff --git a/Sources/SparkConnect/SparkSession.swift
b/Sources/SparkConnect/SparkSession.swift
index 5a007f6..6a366aa 100644
--- a/Sources/SparkConnect/SparkSession.swift
+++ b/Sources/SparkConnect/SparkSession.swift
@@ -608,7 +608,9 @@ public actor SparkSession {
let response = try await session.client.connect(session.sessionID)
await session.setVersion(response.sparkVersion.version)
let isSuccess = try await session.client.setConf(map: sparkConf)
- assert(isSuccess)
+ guard isSuccess else {
+ throw SparkConnectError.InvalidArgument
+ }
return session
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
