HyukjinKwon opened a new pull request, #200: URL: https://github.com/apache/spark-connect-rust/pull/200
### What changes were proposed in this pull request? [SPARK-61000](https://issues.apache.org/jira/browse/SPARK-61000): `DataFrame.toPandas()` now converts each column of the Arrow result by its Spark type, as PySpark's `SparkConnectClient.to_pandas` does. - It first reads three settings in one request: the session time zone, `spark.sql.execution.pandas.structHandlingMode`, and `spark.sql.execution.arrow.pyspark.selfDestruct.enabled`. The `pandasStructHandlingMode` of `_to_pandas` overrides the struct handling mode, as in PySpark. - It converts the result with `pyspark.sql.pandas.conversion._convert_arrow_table_to_pandas`, of the schema of the result that the server sends with it, through the new `DataFrame::collect_record_batches_with_schema` of the Rust core. - The metrics and observed metrics of the execution are in `pdf.attrs`, as in PySpark. So timestamps are naive in the session time zone, maps are dicts, and values of user-defined types are deserialized. `pyspark.sql.pandas.conversion`, vendored from Apache Spark 4.3.0, did not import, so now: - `pyspark/errors/exceptions/captured.py` and `pyspark/traceback_utils.py` are vendored from Apache Spark 4.3.0, and checked by the drift check. `captured.py` imports py4j only where it uses the JVM. - `pyspark/util.py` has PySpark's local socket helpers, verbatim, which it imports. ### Why are the changes needed? `toPandas()` returned `pyarrow.Table.to_pandas()` of the result. Against a Spark 4.3.0 server: | Column | PySpark 4.3.0 and now | Before | | --- | --- | --- | | `TIMESTAMP'2024-01-02 03:04:05'` | `datetime64[ns]`, `Timestamp('2024-01-02 03:04:05')` in the session time zone | `datetime64[us, UTC]`, `Timestamp('2024-01-02 03:04:05+0000', tz='UTC')` | | `map('k', 1)` | `{'k': 1}` | `[('k', 1)]` | | a `VectorUDT` | `DenseVector([1.5, 2.0])` | `{'type': 1, 'size': None, 'indices': None, 'values': array([1.5, 2.])}` | | `pdf.attrs` | `{'metrics': [PlanMetrics, ...]}` | `{}` | ### Does this PR introduce _any_ user-facing change? Yes, `toPandas()` gives the pandas DataFrames of PySpark 4.3.0. ### How was this patch tested? - Against a local Spark 4.3.0 server, `toPandas()` gives the same dtypes and values as official PySpark 4.3.0 for timestamps, NTZ timestamps, structs, maps, arrays, decimals, dates, nulls, binaries and vectors. So do its `attrs`, with and without observed metrics, an empty result, `_to_pandas` with each struct handling mode, and a session time zone of `Asia/Seoul` with a client in `America/Los_Angeles`. Before this change, the timestamps, maps, vectors and `attrs` differed. - A new live test checks these. It also passes with official PySpark 4.3.0, and fails without this change. A new offline test checks that `pyspark.sql.pandas.conversion` imports. - The Arrow handoff test of `toPandas` now expects PySpark's conversion of the same Arrow table, and its mock server serves the settings. - The lib tests and every integration test, e2e included, against the local server (660 passed, and 123 of the transport crate), the build of all tests with `wasm-udf`, the offline Python suites (887 passed, 9 skipped), and the live drop-in and Arrow suites (82 passed) and live cases (10 passed). In the changed lines, clippy reports only the `result_large_err` of the `py.detach` closures, which every action of the Python `DataFrame` has. ### 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]
