This is an automated email from the ASF dual-hosted git repository.
kou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-swift.git
The following commit(s) were added to refs/heads/main by this push:
new 20dac0a fix: Guard against force unwraps on malformed ArrowReader
input (#183)
20dac0a is described below
commit 20dac0a2c4049915eeed09aab13a617763788a55
Author: demetrius albuquerque <[email protected]>
AuthorDate: Thu Sep 3 02:04:02 2026 +0200
fix: Guard against force unwraps on malformed ArrowReader input (#183)
## What's Changed
`readStreaming`, `readFile`, and `fromMessage` force-unwrapped optional
schema and message header values, causing crashes on malformed or
out-of-order input instead of returning an `ArrowError`. For example, a
RecordBatch message arriving with no preceding Schema message would
crash the process rather than fail gracefully.
Replaced the force unwraps with guard statements that return `.failure`
instead:
- `readStreaming`: guards on the RecordBatch header, `schemaMessage`,
and `result.schema` before use
- `readFile`: guards on `footer.schema`, the RecordBatch header, and
`result.schema` before use
- `fromMessage`: guards on the Schema header, the RecordBatch header,
`result.messageSchema`, and `result.schema` before use
While writing a regression test for the above, found that
`readStreaming` also assumed zero-based `Data` indices internally.
Slicing a `Data` (`data[offset...]`) preserves the original absolute
indices rather than resetting to zero, so a caller passing a slice —
rather than a freshly allocated `Data` — could desync the reader's
internal offset tracking and crash on a completely valid, well-formed
input. Fixed by offsetting from `input.startIndex` instead of assuming
0.
## Testing
Added a regression test (`testReadStreamingRecordBatchBeforeSchema`)
that builds a valid streaming payload, strips the Schema message via
slicing, and confirms `readStreaming` now returns `.failure` instead of
crashing. This test only passes with both fixes in place — reverting
either one reproduces a crash.
Full test suite passes (48 tests, 0 failures).
Closes #182
---
Sources/Arrow/ArrowReader.swift | 48 +++++++++++++++++++++++++++++++----------
Tests/ArrowTests/IPCTests.swift | 45 ++++++++++++++++++++++++++++++++++++++
2 files changed, 82 insertions(+), 11 deletions(-)
diff --git a/Sources/Arrow/ArrowReader.swift b/Sources/Arrow/ArrowReader.swift
index d1c0853..ee91eba 100644
--- a/Sources/Arrow/ArrowReader.swift
+++ b/Sources/Arrow/ArrowReader.swift
@@ -274,7 +274,7 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
}
offset += Int(MemoryLayout<UInt32>.size)
- streamData = input[offset...]
+ streamData = input[(input.startIndex + offset)...]
var dataBuffer = ByteBuffer(
data: streamData,
allowReadingUnalignedBuffers: useUnalignedBuffers
@@ -282,11 +282,19 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
let message: org_apache_arrow_flatbuf_Message =
getRoot(byteBuffer: &dataBuffer)
switch message.headerType {
case .recordbatch:
- let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self)!
+ guard let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self) else {
+ return .failure(.invalid("RecordBatch header not found"))
+ }
+ guard let schemaMsg = schemaMessage else {
+ return .failure(.invalid("Schema must be defined before
RecordBatch"))
+ }
+ guard let schema = result.schema else {
+ return .failure(.invalid("Schema not loaded"))
+ }
let recordBatchResult = loadRecordBatch(
rbMessage,
- schema: schemaMessage!,
- arrowSchema: result.schema!,
+ schema: schemaMsg,
+ arrowSchema: schema,
data: input,
messageEndOffset: (Int64(offset) + Int64(length)))
switch recordBatchResult {
@@ -335,7 +343,10 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
data: footerData,
allowReadingUnalignedBuffers: useUnalignedBuffers)
let footer: org_apache_arrow_flatbuf_Footer = getRoot(byteBuffer:
&footerBuffer)
- let schemaResult = loadSchema(footer.schema!)
+ guard let footerSchema = footer.schema else {
+ return .failure(.invalid("Footer schema not found"))
+ }
+ let schemaResult = loadSchema(footerSchema)
switch schemaResult {
case .success(let schema):
result.schema = schema
@@ -368,11 +379,16 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
let message: org_apache_arrow_flatbuf_Message =
getRoot(byteBuffer: &mbb)
switch message.headerType {
case .recordbatch:
- let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self)!
+ guard let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self) else {
+ return .failure(.invalid("RecordBatch header not found"))
+ }
+ guard let schema = result.schema else {
+ return .failure(.invalid("Schema not loaded"))
+ }
let recordBatchResult = loadRecordBatch(
rbMessage,
- schema: footer.schema!,
- arrowSchema: result.schema!,
+ schema: footerSchema,
+ arrowSchema: schema,
data: fileData,
messageEndOffset: messageEndOffset)
switch recordBatchResult {
@@ -421,7 +437,9 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
let message: org_apache_arrow_flatbuf_Message = getRoot(byteBuffer:
&mbb)
switch message.headerType {
case .schema:
- let sMessage = message.header(type:
org_apache_arrow_flatbuf_Schema.self)!
+ guard let sMessage = message.header(type:
org_apache_arrow_flatbuf_Schema.self) else {
+ return .failure(.invalid("Schema header not found"))
+ }
switch loadSchema(sMessage) {
case .success(let schema):
result.schema = schema
@@ -431,9 +449,17 @@ public class ArrowReader { // swiftlint:disable:this
type_body_length
return .failure(error)
}
case .recordbatch:
- let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self)!
+ guard let rbMessage = message.header(type:
org_apache_arrow_flatbuf_RecordBatch.self) else {
+ return .failure(.invalid("RecordBatch header not found"))
+ }
+ guard let messageSchema = result.messageSchema else {
+ return .failure(.invalid("Schema must be defined before
RecordBatch"))
+ }
+ guard let schema = result.schema else {
+ return .failure(.invalid("Schema not loaded"))
+ }
let recordBatchResult = loadRecordBatch(
- rbMessage, schema: result.messageSchema!, arrowSchema:
result.schema!,
+ rbMessage, schema: messageSchema, arrowSchema: schema,
data: dataBody, messageEndOffset: 0)
switch recordBatchResult {
case .success(let recordBatch):
diff --git a/Tests/ArrowTests/IPCTests.swift b/Tests/ArrowTests/IPCTests.swift
index 58f973e..7d7e972 100644
--- a/Tests/ArrowTests/IPCTests.swift
+++ b/Tests/ArrowTests/IPCTests.swift
@@ -262,6 +262,50 @@ final class IPCStreamReaderTests: XCTestCase {
throw error
}
}
+
+ func testReadStreamingRecordBatchBeforeSchema() throws {
+ // Build a minimal streaming message: a RecordBatch header with no
+ // preceding Schema message. This should fail gracefully instead
+ // of crashing on a force unwrap.
+ let schema = makeSchema()
+ let recordBatch = try makeRecordBatch()
+ let arrowWriter = ArrowWriter()
+ let writerInfo = ArrowWriter.Info(.recordbatch, schema: schema,
batches: [recordBatch])
+
+ switch arrowWriter.writeStreaming(writerInfo) {
+ case .success(let writeData):
+ // Mirror the parsing logic in ArrowReader.readStreaming to advance
+ // past exactly one message (the Schema message), leaving the
+ // RecordBatch message intact and correctly positioned for
+ // readStreaming to parse on its own.
+ var offset = 0
+ var length = getUInt32(writeData, offset: offset)
+ if length == CONTINUATIONMARKER {
+ offset += Int(MemoryLayout<UInt32>.size)
+ length = getUInt32(writeData, offset: offset)
+ }
+ offset += Int(MemoryLayout<UInt32>.size)
+
+ var dataBuffer = ByteBuffer(
+ data: writeData[offset...],
+ allowReadingUnalignedBuffers: false)
+ let message: org_apache_arrow_flatbuf_Message =
getRoot(byteBuffer: &dataBuffer)
+ XCTAssertEqual(message.headerType, .schema)
+
+ offset += Int(message.bodyLength + Int64(length))
+ let truncatedData = writeData[offset...]
+
+ let arrowReader = ArrowReader()
+ switch arrowReader.readStreaming(truncatedData) {
+ case .success:
+ XCTFail("Expected failure when RecordBatch precedes Schema")
+ case .failure:
+ break // Correct: should fail gracefully, not crash
+ }
+ case .failure(let error):
+ throw error
+ }
+ }
}
final class IPCFileReaderTests: XCTestCase { // swiftlint:disable:this
type_body_length
@@ -671,5 +715,6 @@ final class IPCFileReaderTests: XCTestCase { //
swiftlint:disable:this type_body
throw error
}
}
+
}
// swiftlint:disable:this file_length
