Skip to content

Comet Native Executors: partition hash exchanges with Spark's Murmur3 hash #7368

Open

Description

@andygrove

What is the problem the feature request solves?

On a Comet Native Executor cluster (#7289), a hash exchange becomes a RepartitionExec(Hash), which Ballista shuffles with DataFusion's hash function rather than Spark's Murmur3. That is consistent within a query, which is all joins, aggregates and windows need. A plan that relies on Spark's own placement of rows, such as a join with a bucketed table read without a shuffle, or a storage-partitioned join, would return wrong results.

Describe the potential solution

apache/datafusion-ballista#2723, on Ballista's comet-native branch, lets a hash shuffle place rows by a partition id the client computes into a column. The producing stage could compute Spark's partition id with Comet's own CometPartitioning hash, and the exchange would route by it. Ballista's adaptive planner currently refuses that partition-id marker, so this also needs that path supported there.

Additional context

The hash exchange is built in exchange in native/native-executors/src/assembly.rs, whose comment notes the limitation.

Part of #7288.

Willingness to contribute

Yes. I can contribute this feature independently.

Metadata

Metadata

Assignees

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions