Remove the hash-partitioning shuffle writer

[andygrove]

Which issue does this PR close?

N/A

Rationale for this change

Ballista had two shuffle writer implementations selected by a session config
(ballista.shuffle.sort_based.enabled, default true):

• Sort-based (SortShuffleWriterExec) — writes one consolidated data file plus an
  index file per input partition (2 x N files), coalesces small batches, and bounds
  memory via spilling.
• Hash-based (ShuffleWriterExec's hash path) — keeps one open Arrow StreamWriter
  per output partition and writes one file per (input_partition, output_partition)
  pair (N x M files). This per-output-partition state does not scale to high output
  partition counts.

Sort-based has been the default and supersedes the hash path for the case that matters
(hash-repartition stages). Keeping the non-scaling hash writer and a config toggle to
select it added complexity and a foot-gun without a real use case. This PR removes it.

What changes are included in this PR?

• Planner: hash-repartition stages (Partitioning::Hash) now always use
  SortShuffleWriterExec; the ballista.shuffle.sort_based.enabled gate is gone.
  create_shuffle_writer_with_config errors on any other Some(..) partitioning
  instead of silently building a writer that could not be encoded.
• Writer: removed ShuffleWriterExec's hash write path (the per-output-partition
  WriteTracker array + BatchPartitioner) and the unused repart_time metric.
  ShuffleWriterExec is retained for single-partition (None) output — the final
  output stage, CoalescePartitionsExec, SortPreservingMergeExec, and broadcast-build
  stages. Since it no longer repartitions at all, its Option<Partitioning> is gone
  outright: try_new drops the argument, the struct drops the field and derives its
  properties from the child plan, and the ShuffleWriter trait impl returns None
  unconditionally. ShuffleWriterVariant::Hash is renamed to Passthrough to match.
• Reader / Flight: sort-shuffle reads are gated purely on the on-disk index file
  (is_sort_shuffle_output), which was already the authoritative signal; the redundant
  config-derived boolean and the ballista_sort_shuffle_enabled trait method are removed.
• Config: removed the ballista.shuffle.sort_based.enabled key, its ConfigEntry,
  and its accessor. Sort-shuffle tuning keys (batch_size,
  memory_limit_per_task_bytes, writer channel capacity) are unchanged.
• Benchmark: shuffle_bench is now sort-only (dropped --writer hash).
• Tests: removed the hash-vs-sort comparison tests and the sort-shuffle enable-toggle
  tests; adapted writer/reader unit tests that constructed hash-partitioned writers to the
  single-partition path.
• Docs: removed the "Hash-based shuffle (opt-in)" documentation and the
  sort_based.enabled config rows.

TPC-H plan-stability golden files are unchanged: they already show
SortShuffleWriterExec for hash-repartition stages and ShuffleWriterExec: partitioning: None for terminal stages, so query plan shape does not move under the default
configuration.

Are there any user-facing changes?

Yes — breaking. The ballista.shuffle.sort_based.enabled config key and the public
SessionConfigExt::ballista_sort_shuffle_enabled trait method are removed, and the
hash-based shuffle writer is no longer available. BallistaConfig does not reject unknown
keys, so a config that still sets the removed key will be accepted and silently ignored
(sort-based shuffle is used unconditionally) rather than erroring.

Two further public API breaks for downstream projects that build plans directly:
ShuffleWriterExec::try_new loses its trailing shuffle_output_partitioning argument,
and ShuffleWriterVariant::Hash is renamed to ShuffleWriterVariant::Passthrough.
The api change label is applied.

Note on wire compatibility: scheduler and executor deploy together and serialized plans
are ephemeral, but for completeness — a legacy physical plan carrying a hash
output_partitioning inside a ShuffleWriterExecNode now fails to decode, since
hash-repartition stages are always emitted as SortShuffleWriterExec.

[milenkovicm]

should we merge this after 54.1 has been released (if we assume releasing current main) as it will bring breaking change ?

[andygrove]

@phillipleblanc could you help review?

[andygrove]

should we merge this after 54.1 has been released (if we assume releasing current main) as it will bring breaking change ?

I'm planning on cherry picking commits to the 54 release branch

[avantgardnerio]

I think this conflicts with the direction I was heading. I'm "requesting changes" so I have a chance to look it over even if someone else approves. I'll respond with real feedback shortly.

[avantgardnerio]

I see this removes ShuffleWriter(Hash) which I think is a good thing. I see it doesn't remove ShuffleWriter(Passthrough) which was my main concern. I'm about to start adding RangeRepartitionExec and that will require the passthrough mode. So I think my biggest concern is addressed. Looking over the rest now.

[avantgardnerio]

Changing to "comment" so I don't hold up the works, now that I know my critical concern was a non-issue.

Trying to retract my change request.

[avantgardnerio]

I'm considering case B here, now that we have multi-partition-per-task, there's some instances where it could be better to hash parition:

  Case A: 1 input partition per task (4 tasks total)
  - Sort-shuffle: 4 tasks × 1 input each = 4 files total. Downstream reader for output k opens 4 files, seeks to index[k] in each.
  - Hash-writer (M×K layout): 4 tasks × 4 outputs each = 16 files total. Downstream reader for output k opens 4 files (one per producer task), reads each whole.
  
  Case B: 4 input partitions per task (1 task total, the multi-partition-tasks payoff)
  - Sort-shuffle: 1 task × 4 inputs = 4 files total. Downstream reader for output k still opens 4 files with 4 index seeks — same file count as Case A.
  - Hash-writer: 1 task × 4 outputs = 4 files total. Downstream reader for output k opens 1 file, reads whole.

But what I was working towards anyway was removing that hashing from ShuffleWriter entirely:

RepartitionExec(Hash) → ShuffleWriterExec(None)

Which I think this PR still allows.

[andygrove]

I'm considering case B here, now that we have multi-partition-per-task, there's some instances where it could be better to hash parition:

  Case A: 1 input partition per task (4 tasks total)
  - Sort-shuffle: 4 tasks × 1 input each = 4 files total. Downstream reader for output k opens 4 files, seeks to index[k] in each.
  - Hash-writer (M×K layout): 4 tasks × 4 outputs each = 16 files total. Downstream reader for output k opens 4 files (one per producer task), reads each whole.
  
  Case B: 4 input partitions per task (1 task total, the multi-partition-tasks payoff)
  - Sort-shuffle: 1 task × 4 inputs = 4 files total. Downstream reader for output k still opens 4 files with 4 index seeks — same file count as Case A.
  - Hash-writer: 1 task × 4 outputs = 4 files total. Downstream reader for output k opens 1 file, reads whole.

But what I was working towards anyway was removing that hashing from ShuffleWriter entirely:

RepartitionExec(Hash) → ShuffleWriterExec(None)

Which I think this PR still allows.

If there are valid cases for keeping hash-based partition where there are tiny number of output partitions, we can keep it in. I am targeting use cases that could have very high number of output partitions (hundreds/thousands).

[avantgardnerio]

I was also looking to remove ShuffleWriter(Hash), but was looking at this from a slightly different perspective: I'd love to remove all shuffling from writers and only have ShuffleWriter(None) (I'm not sure if this is feasible or not yet for SortShuffleWriter.

The end goal for me is something like:

ArbitraryPartitioner -> ShuffleWriter(Passthrough)

Which allows for the RangeRepartitioning I was about to put up a PR for. I think this PR preserves that ability, but I'd like to explicitly call it out, and even go so far as to say standard hash partitioning is better for some cases where partitions_per_task > 1 (see comments for an example).

It's possible that we'll hit some friction down the road by conflating partitioning with writing, but we can deal with that when we get there (shortly). On the whole I think this PR is a step in the right direction, by deleting the Hash branch.

[avantgardnerio]

very high number of output partitions (hundreds/thousands).

Let's say there's a company with a 250 node cluster, each with 32 vcores. Currently, with paritions_per_task = 32, that would create 8000 sort shuffle output files. Stage N+1 would need to read a small chunk of all 8000 files. Compare that with Hash partitioning, which would output 250 files. I believe this table is correct:

IIUC, that's a manageable number of files, but a very high amount of network requests.

[avantgardnerio]

To put this more simply:

N*M has turned into N*M/vcores but only for Hash partitioning, and the denominator can be significant (we have up to 45 cores). But it doesn't apply to SortShuffleWriter and it would be interesting to see if we could make it apply, longer term.

[phillipleblanc]

Looks good to me - we should probably put the api change label on this PR since its a breaking change.

[andygrove]

Thanks for the reviews/feedback. I will aim to address feedback tomorrow.

[andygrove]

@avantgardnerio @phillipleblanc I addressed feedback. PTAL when you can.

[avantgardnerio]

@avantgardnerio @phillipleblanc I addressed feedback. PTAL when you can.

I stand by my previous ✔️ . Thanks for doing this! It's funny we were both working on the same thing at the same time.