Tim Meehan created SPARK-61285:
----------------------------------
Summary: Partition by range in DataFrame::repartition_by_range of
the Rust API, as PySpark's repartitionByRange does
Key: SPARK-61285
URL: https://issues.apache.org/jira/browse/SPARK-61285
Project: Spark
Issue Type: Bug
Components: Connect
Affects Versions: connect-rust-4.3.0
Reporter: Tim Meehan
DataFrame::repartition_by_range and repartition_by_range_opt of the Rust API
send their columns as they are. The server range-partitions only on sort
orders, so a plain column is hash-partitioned: the request is the one that
repartition(4, col("id")) sends. PySpark's repartitionByRange wraps each column
that is not a sort order in an ascending one, and so does the Python client of
spark-connect-rust.
{code:rust}
// spark is a SparkSession of the spark-connect crate.
let df = spark.range(100)?.repartition_by_range(4,
vec![col("id").expression().clone()]);
{code}
The min, max and count of id per partition, against an OSS Spark 4.3.0 Connect
server on loopback:
* repartitionByRange(4, "id") of pyspark-client 4.3.0, and of the Python client
at spark-connect-rust master 8768cc4: (0, 24, 25), (25, 49, 25), (50, 74, 25),
(75, 99, 25).
* repartition(4, "id"), the request the Rust API sends: (2, 93, 28), (0, 99,
24), (14, 94, 24), (3, 98, 24).
The rows are the same, but not which rows share a partition, so the files that
a write produces differ, and a sortWithinPartitions afterwards does not sort
globally.
>From source, at 8768cc4: LogicalPlan::RepartitionByRange pushes each
>expression unchanged into partition_exprs of RepartitionByExpression
>(crates/spark-connect/src/plan.rs:1612-1628).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
