andygrove commented on code in PR #2106:
URL:
https://github.com/apache/datafusion-ballista/pull/2106#discussion_r3623840132
##########
ballista/core/src/execution_plans/shuffle_writer.rs:
##########
@@ -456,6 +436,14 @@ impl ShuffleWriterExec {
work_dir: String,
shuffle_output_partitioning: Option<Partitioning>,
) -> Result<Self> {
+ if matches!(shuffle_output_partitioning, Some(Partitioning::Hash(..)))
{
Review Comment:
Done in beee6398. You were right that the runtime guard was the wrong shape.
The field was provably dead: the planner passes `None` at all four call sites,
the AQE adapter only ever passes `None` or `Hash` (and `Hash` routes to
`SortShuffleWriterExec`), and proto decode can only produce `None` or `Hash`
since `parse_protobuf_hash_partitioning` yields nothing else. So `try_new` and
the struct both lose the `Option<Partitioning>`, and the trait impl returns
`None` unconditionally.
The one thing that survives is the enum in `execution_engine.rs`, since the
two writers are distinct concrete types and the variant is what tells the
executor which on-disk layout to stamp onto summaries. But `Hash` was an
actively wrong name for it, so it is now `Passthrough`.
One behavior change worth flagging: `create_shuffle_writer_with_config` now
errors on a non-hash `Some(..)` rather than falling through. That case used to
build a writer that would fail later at proto encode, so this just moves the
failure to plan time.
##########
ballista/scheduler/src/planner.rs:
##########
@@ -700,36 +700,33 @@ pub(crate) fn create_shuffle_writer_with_config(
) -> Result<Arc<dyn ShuffleWriter>> {
let plan = make_empty_exec_serde_safe(plan)?;
- // Check if sort-based shuffle is enabled
+ // Sort-based shuffle is the only shuffle writer for hash-repartition
+ // stages. Its tuning values still come from the session config.
let ballista_config = config
.extensions
.get::<BallistaConfig>()
.cloned()
.unwrap_or_default();
- if ballista_config.shuffle_sort_based_enabled() {
- // Sort shuffle requires hash partitioning
- if let Some(Partitioning::Hash(exprs, partition_count)) = partitioning
{
- let sort_config = SortShuffleConfig::new(
- true,
- ballista_config.shuffle_sort_based_batch_size(),
- )
- .with_memory_limit_per_task_bytes(
-
ballista_config.shuffle_sort_based_memory_limit_per_task_bytes(),
- );
+ // Sort shuffle requires hash partitioning.
+ if let Some(Partitioning::Hash(exprs, partition_count)) = partitioning {
+ let sort_config =
+ SortShuffleConfig::new(true,
ballista_config.shuffle_sort_based_batch_size())
+ .with_memory_limit_per_task_bytes(
+
ballista_config.shuffle_sort_based_memory_limit_per_task_bytes(),
+ );
- return Ok(Arc::new(SortShuffleWriterExec::try_new(
- job_id.to_owned(),
- stage_id,
- plan,
- "".to_owned(),
- Partitioning::Hash(exprs, partition_count),
- sort_config,
- )?));
- }
+ return Ok(Arc::new(SortShuffleWriterExec::try_new(
+ job_id.to_owned(),
+ stage_id,
+ plan,
+ "".to_owned(),
+ Partitioning::Hash(exprs, partition_count),
+ sort_config,
+ )?));
}
- // Fall back to standard shuffle writer
+ // Single-partition / non-hash stages use the single-file writer.
Review Comment:
Fixed, it now reads "Stages that don't repartition write their input
partitioning through: one file per output partition."
--
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]
