HyukjinKwon opened a new pull request, #264: URL: https://github.com/apache/spark-connect-rust/pull/264
### What changes were proposed in this pull request? [SPARK-61056](https://issues.apache.org/jira/browse/SPARK-61056): `df.writeStream.foreach(f)` now sends the function of PySpark, so the queries of `foreach` run. As PySpark Connect's, `foreach(f)` sends the function of a partition and its rows of `_construct_foreach_function(f)`, which the Python worker of the server runs: - for a function `f`, one that calls `f` with each row - for an object, one that calls its `open(partition_id, epoch_id)`, its `process(row)` with each row, and its `close(error)` It raises `ATTRIBUTE_NOT_CALLABLE` for an object without a callable `process`, or with an `open` or `close` that is not callable. The function is ported from PySpark, but it imports `TaskContext` on the worker, which has it. `foreach` and `foreachBatch` now pickle their functions with `CloudPickleSerializer`, as PySpark's. They raise `STREAMING_CONNECT_SERIALIZATION_ERROR` for a function that does not pickle. ### Why are the changes needed? The client sent `f` itself, which the worker calls with a partition and its rows: - the query of an object failed with `TypeError: 'Writer' object is not callable` - that of a function failed without a batch So `foreach` did not work. ### Does this PR introduce _any_ user-facing change? Yes. The queries of `foreach` run, as in PySpark 4.3.0. ### How was this patch tested? - A new live test runs `foreach` of a function and of an object of `open`, `process` and `close`, which write a file of each row on the local server, and checks the files. It also checks the `ATTRIBUTE_NOT_CALLABLE` of an int. It passes with official PySpark 4.3.0, and fails without this change. - A new offline test unpickles the command that `foreach` sends to a mock Connect server, runs its function on rows, and checks the errors of objects without callable methods and of functions that do not pickle. - The lib tests of both crates (577 passed), the build of all tests, also with `wasm-udf`, the offline Python suites (1817 passed, 8 skipped), and the live drop-in and Arrow suites (116 passed, without the test of `StreamingQuery.explain`, which hangs on a slow machine and #260 fixes) and live cases (10 passed) against a local Spark 4.3.0 server. Clippy reports no diagnostics in the changed lines. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 This pull request and its description were written by Isaac. -- 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]
