Replace / rename HashPartitioned distribution as KeyPartitioned

[gene-bordegaray]

Is your feature request related to a problem or challenge?

Distribution::HashPartitioned is documented as requiring rows with equal key values to land in the same partition. While working through range partitioning, @2010YOUY01 and @gabotechs pointed out this is really a key-partitioning contract, not a requirement that the existing input is specifically hash partitioned.

This name has historically caused confusion / misuse, and as range partitioning support expands this continues to come up. The key point is that range partitioning can satisfy some single-input key partitioning requirements without specifically being hash partitioned.

Describe the solution you'd like

Clarify the API direction for this distribution requirement. Options discussed include:

• keep HashPartitioned but document it as historical naming for key partitioning (I am not a fan of this one)
• migrate to a KeyPartitioned name and have both Partitioning::Hash and compatible Partitioning::Range satisfy this

This issue is only about the per-input distribution requirement. Multi-input / join co-partitioning should be handled separately.

NOTE: I would prefer this rename / replacement to happen after aggregations (a unary operator) and joins (a multi-input operator) HashPartitioned distributions are satisfied via range partitioning before replacing the public HashPartitioned variant. I want to take this approach to ensure we have worked out a majority of the kinks and nuances of this replacement before making large public API changes.

Additional context

Epic: #22395

Related PRs / discussion:

• Allow Range partitioning to satisfy grouped aggregation requirements #23191
• Support co-partitioned range inner equi joins #23184

[gene-bordegaray]

also @stuhood @saadtajwar @alamb

[alamb]

keep HashPartitioned but document it as historical naming for key partitioning (I am not a fan of this one)

I vote we start with this one, as it is clearly better than the status quo and has no downstream API implications

migrate to a KeyPartitioned name and have both Partitioning::Hash and compatible Partitioning::Range satisfy this

This seems like a good idea on the surface, but I think we would have to evaluate how disruptive it could be on on other DataFusion users

I am not really sure how to do this other than to try it out with a prototype

[gabotechs]

I'm trying to look in practice what would be the blast radius of a plain rename here:

• [draft] Rename Distribution::HashPartitioned to Distribution::KeyPartitioned #23241

Still trying it out, but so far it looks like the blast radius is pretty contained.

[gene-bordegaray]

keep HashPartitioned but document it as historical naming for key partitioning (I am not a fan of this one)

I vote we start with this one, as it is clearly better than the status quo and has no downstream API implications

This is what I started with here and agree for the short-term. I meant long-term this name should probably be changed as its a bit misleading.

[gabotechs]

Did an analysis here about the blast radius of changing the name #23241. The blast radius is not huge, but there do are some downstream crates relying on it (e.g. InfluxDB, Ballista, GreptimeDB, delta-rs, ...)

[gene-bordegaray]

What do we think about updating the documentation on it as historical naming and deperecating it with a.warning this will be replaced with KeyPartitioning and eventually not be supported.

Then we can use the HashPartitioned internally for range right now while dong the first unary (aggregates) and multi-input (joins) to ensure the semantics are correct and allow others to chime in on their thoughts before public decision.

Finally can do a deprecating change in a separate PR.

[alamb]

What do we think about updating the documentation on it as historical naming and deperecating it with a.warning this will be replaced with KeyPartitioning and eventually not be supported.

I think that would be the most friendly way

So basically in the context of #23241 it would look like

/// How data is distributed amongst partitions. See [`Partitioning`] for more
/// details.
#[derive(Debug, Clone)]
pub enum Distribution {
    /// Unspecified distribution
    UnspecifiedDistribution,
    /// A single partition is required
    SinglePartition,
    /// Comments  here about deprecated
    #[deprecated] 
    HashPartitioned(Vec<Arc<dyn PhysicalExpr>>), 
    /// Requires children to be distributed in such a way that the same
    /// values of the keys end up in the same partition
    KeyPartitioned(Vec<Arc<dyn PhysicalExpr>>),
}

And then all the match statements owuld look like

match dist { 
  Distribution::UnspecifiedDistribution => {..}
  Distribution::SinglePartition => {..}
  Distribution::HashPartitioned(exprs) |  Distribution::KeyPartitioned(exprs) => { /* treat both the same ... */ }
}

[gene-bordegaray]

/// How data is distributed amongst partitions. See [Partitioning] for more
/// details.
#[derive(Debug, Clone)]
pub enum Distribution {
/// Unspecified distribution
UnspecifiedDistribution,
/// A single partition is required
SinglePartition,
/// Comments here about deprecated
#[deprecated]
HashPartitioned(Vec<Arc>),
/// Requires children to be distributed in such a way that the same
/// values of the keys end up in the same partition
KeyPartitioned(Vec<Arc>),
}
And then all the match statements owuld look like

match dist {
Distribution::UnspecifiedDistribution => {..}
Distribution::SinglePartition => {..}
Distribution::HashPartitioned(exprs) | Distribution::KeyPartitioned(exprs) => { /* treat both the same ... */ }
}

I am good with this, especially since internal blast radius is small

@gabotechs @2010YOUY01 what do you guys think

[gabotechs]

Sounds good, let's do that.

[2010YOUY01]

And then all the match statements owuld look like

match dist {
Distribution::UnspecifiedDistribution => {..}
Distribution::SinglePartition => {..}
Distribution::HashPartitioned(exprs) | Distribution::KeyPartitioned(exprs) => { /* treat both the same ... */ }
}

I think this is a good idea.