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

Labels

enhancementNew feature or request

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions