andygrove opened a new issue, #2681: URL: https://github.com/apache/datafusion-ballista/issues/2681
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** A hash shuffle always assigns rows to partitions with DataFusion's hash. `compute_partition_indices` in the sort-shuffle writer hashes the keys with `create_hashes` and `REPARTITION_RANDOM_STATE` ([sort_shuffle/writer.rs#L1393-L1415](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/core/src/execution_plans/sort_shuffle/writer.rs#L1393-L1415)), byte-identical to DataFusion's `BatchPartitioner`. That is consistent within a job, so joins and aggregates over shuffled inputs are correct. What it can't do is match data that is already partitioned by another function. Spark partitions by Murmur3 (seed 42, then `pmod`), and bucketed tables and storage-partitioned joins depend on that. When one side of a join is a Spark-bucketed table read without a shuffle, the other side has to be shuffled with Spark's hash for matching rows to meet. Comet Native Executors (apache/datafusion-comet#7288) turn bucketing off for this reason. Spark's placement of rows also decides which output file each row goes to in a `repartition(n, col)` write. **Describe the solution you'd like** The smallest general option is to let the hash shuffle writer route rows by a partition-id column that the plan computes, instead of hashing the keys itself. A client then computes whatever placement it needs in its own operator below the writer (for Comet, Spark's `pmod(murmur3(keys), n)`), and the writer drops the column before writing. This could be a new partitioning mode in Ballista's shuffle writer, planned from a marker node. Another option is a pluggable hasher: an `Arc<dyn ...>` that maps a batch's key columns to hashes, registered in the executor's config, with an identifier carried in the plan so that every task of a job uses the same one. That keeps hashing in the writer, but every executor needs the hasher. **Describe alternatives you've considered** - Computing Spark's partition id into a column and hash-partitioning on that column. Rows with the same Spark partition still meet, but Ballista's partition `i` doesn't hold Spark's partition `i`, so the result can't line up with bucketed data. - A pluggable shuffle writer and reader, which #1539 asks for to support a remote shuffle service. That would also allow this, at a much larger cost. **Additional context** Expressions inside Ballista's shuffle nodes are encoded with `DefaultPhysicalExtensionCodec`, not the codec a client configures, so a key that is a client's own `PhysicalExpr` can't be sent to executors today. A partition-id column avoids needing that. Part of apache/datafusion-comet#7288. -- 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]
