HyukjinKwon opened a new pull request, #210: URL: https://github.com/apache/spark-connect-rust/pull/210
### What changes were proposed in this pull request? [SPARK-61007](https://issues.apache.org/jira/browse/SPARK-61007): the Python client now supports PySpark's subquery expressions. - `DataFrame.scalar()` and `DataFrame.exists()` return the Columns of a scalar and an EXISTS subquery of the DataFrame. - `Column.isin(df)` returns that of an IN subquery, of the fields of a `struct(...)` for one. - A UDTF call takes the `TableArg` of `DataFrame.asTable()`, partitioned and ordered as given, as an argument that passes the rows of the DataFrame. These are `SubqueryExpression`s of the plan of the DataFrame, by its plan ID. As PySpark's `LogicalPlan._with_relations` does, a relation whose expressions have subqueries is sent in a `WithRelations` with the plans that they refer to: ``` with_relations [plan ID of the relation] root: the relation [its root plan ID] references: the plan of each subquery [its plan ID] ``` Each relation collects the subqueries of its own expressions as it is serialized, so this holds for every operator, and for nested subqueries, which are sent in the relations that have them. The plan IDs stay stable, as for any plan. As PySpark's `TableArg`, `partitionBy()`, `orderBy()` and `withSinglePartition()` in the wrong order raise `IllegalArgumentException`. In the core, `DataFrame::scalar()` and `DataFrame::exists()` return the Columns too, as PySpark's do, and `Column::isin_subquery` is the IN subquery. This is based on #209, whose commits are the first ones here. ### Why are the changes needed? `scalar()` ran the query and returned its first value, and `exists()` returned a bool, so the subqueries of PySpark failed. For example, `t1.where(t2.where(col("t2.a") == col("t1.a").outer()).exists())` raised `UNRESOLVED_COLUMN`. `isin(df)` raised `UNSUPPORTED_DATA_TYPE`, and so did a UDTF call with a `TableArg`. ### Does this PR introduce _any_ user-facing change? Yes. `scalar()`, `exists()`, `isin(df)` and `asTable()` work as in PySpark 4.3.0. In the Rust API, `DataFrame::scalar()` and `DataFrame::exists()` now return Columns of the subqueries. ### How was this patch tested? - A new live test runs scalar, correlated and IN subqueries, an IN subquery of a struct, and a UDTF call with the rows of a single, ordered partition and with a named table argument. It also checks the errors of the `TableArg` builders. It passes with official PySpark 4.3.0, and fails without this change. - A new offline test checks, against a mock Connect server, the `WithRelations` of a relation with subqueries, its plan IDs, and the subqueries and table argument that it sends. It fails without this change. - New lib tests check the relations of subqueries, nested ones included, the stability of their plan IDs, and their rendering. A new e2e test checks the rows of subqueries from Rust. The tests of the `TableArg` builders and of `scalar()` and `exists()` now expect PySpark's. - The lib tests and every integration test, e2e included, against a local Spark 4.3.0 server (669 passed, and 125 of the transport crate), the build of all tests with `wasm-udf`, the offline Python suites (1162 passed, 8 skipped), and the live drop-in and Arrow suites (89 passed) and live cases (10 passed). 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]
