dongjoon-hyun opened a new pull request, #618: URL: https://github.com/apache/spark-connect-swift/pull/618
### What changes were proposed in this pull request? This PR fixes `DataFrame` actions to throw `SparkConnectError.unsupportedOperation` instead of returning wrong values or crashing when the Spark Connect server sends compressed Arrow batches. - `ArrowReader` returns an error for a record batch with `compression` instead of reading the compressed buffers as they are. - `DataFrame` actions check the header of each Arrow batch first and throw `unsupportedOperation` with the codec name and the workaround, `spark.sql.execution.arrow.compression.codec=none`. ### Why are the changes needed? Since Spark 4.1.0, the server compresses the buffers of Arrow batches when [spark.sql.execution.arrow.compression.codec](https://github.com/apache/spark/blob/v4.1.0/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4046) is `zstd` or `lz4`. [ArrowBatchIterator](https://github.com/apache/spark/blob/v4.1.0/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala#L118) of `v4.1.0` (2025-12-11) passes the codec to `VectorUnloader`. The server then sets `compression` on every record batch, including small ones whose buffers are stored uncompressed after an 8-byte length of `-1`. However, `ArrowReader` ignores `compression` and reads the compressed buffers as they are. ```swift try await spark.conf.set("spark.sql.execution.arrow.compression.codec", "zstd") try await spark.range(10).count() // -1 try await spark.range(10).collect() // Wrong values like -1 try await spark.range(3).show() // Fatal error: UnsafeBufferPointer with negative count ``` Like the Scala client, which fails with `IllegalArgumentException: Please add arrow-compression module to use CommonsCompressionFactory for ZSTD`, this PR does not support decompression. PySpark decompresses the batches with PyArrow. ### Does this PR introduce _any_ user-facing change? Yes. With Spark 4.1+ and a compression codec, `collect()`, `collect(as:)`, `head()`, `first()`, `take()`, `tail()`, `count()`, and `show()` throw the following error instead of returning wrong values or crashing. Results without buffers, e.g., `df.select()`, also throw it. ``` unsupportedOperation: Unsupported Arrow compression codec: zstd. Set `spark.sql.execution.arrow.compression.codec` to `none`. ``` There is no change when the codec is `none`, which is the default. ### How was this patch tested? Added `RecordBatchTests.readCompressed`, which does not require a Spark Connect server, and `DataFrameTests.arrowCompression` for Spark 4.1+. Verified against an `apache/spark:4.2.0` Connect server. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
