avantgardnerio commented on code in PR #2106:
URL:
https://github.com/apache/datafusion-ballista/pull/2106#discussion_r3615933895
##########
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:
nit: it's no longer a singe file writer, it's the N partition writer
##########
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:
Do we just want to remove the Enum entirely?
--
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]
