LuciferYang opened a new pull request, #59200: URL: https://github.com/apache/spark/pull/59200
### What changes were proposed in this pull request? Backport of #55410 to branch-4.3. The fix binds the Spark Connect Python `foreachBatch` batch DataFrame to the streaming query's own cloned session instead of the root session, so the batch runs under the configuration the streaming engine pins on the clone. A new SQL config `spark.sql.connect.streaming.foreachBatch.useClonedSession` (default `true`) gates it; set it to `false` to restore the old root-session behavior. See #55410 for the full mechanism. One change for branch-4.3: `PythonWorkerEnvironment` does not exist on this branch (it was added later and is unrelated to this fix), so the server keeps branch-4.3's `runner.init()` worker startup instead of the `PythonWorkerEnvironment.readValidated` form on master. Every API the fix needs is present on branch-4.3. The migration guide adds only the "4.2 to 4.3" entry. ### Why are the changes needed? `StreamExecution` runs the query on a clone of the session that started it and pins the configs stateful correctness depends on: CBO off, AQE off for continuous/real-time/stateful workloads, the stateful shuffle partition count fixed at start, and the checkpointed conf restored from the offset log. Connect resolved the batch DataFrame against the root `SessionHolder`, so work the user function did on it ran under the root session's configuration with none of those in effect. The Python worker also kept the root session active, so the Connect session mapping never expired and the query kept running after the client left. branch-4.3 should carry the same fix. ### Does this PR introduce _any_ user-facing change? Yes, the same change as #55410. With the config at its default `true`, the batch DataFrame is bound to the cloned session, matching classic `foreachBatch`. As a result `SparkSession.active()` inside the function returns the batch session, a temporary view created from the batch DataFrame is not visible to the root session, referencing it from `spark.sql` on a captured root session fails with `DATAFRAME_NOT_FOUND`, and joining it with a root-session DataFrame fails with `SESSION_NOT_SAME`. Run such work through `batch_df.sparkSession`, and pass results out through a table or a global temporary view. Set the config to `false` to restore the previous behavior. The Structured Streaming migration guide documents this. ### How was this patch tested? Pass GitHub Actions. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code -- 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]
