MSQ: Use full parallelism in localSort. (#18765)

Since localSort was introduced for sort-merge join in #13506, it has
used 1 processor at a time with a max of 2 channels per processor.
This is inefficient, because at the time the sorter runs, it is the
only sorter running. The patch contains a comment describing the
situation in more detail.

This patch has two benefits. First, the sorter should run faster
if multiple processing threads are available. Second, the sorter,
due to having a larger max channels per merger, will make more
efficient use of intermediate channels.

Author: gianm

multi-stage-query/src/main/java/org/apache/druid/msq/exec/std/StandardShuffleOperations.java

@@ -364,6 +364,7 @@ public boolean isBuffered()
             nextFuture = Futures.transformAsync(
                 nextFuture,
                 ignored -> {
+                  final WorkerMemoryParameters memoryParameters = executionContext.frameContext().memoryParameters();
                   final SuperSorter sorter = new SuperSorter(
                       Collections.singletonList(channel.getReadableChannel()),
                       stageDefinition.getFrameReader(),
@@ -382,8 +383,18 @@ public <T2> FrameProcessor<T2> decorate(FrameProcessor<T2> processor)
                       executionContext.makeIntermediateOutputChannelFactory(
                           StringUtils.format("hash-parts-super-sort-%06d", channel.getPartitionNumber())),
                       executionContext.frameContext().frameWriterSpec().getRowBasedFrameType(),
-                      1,
-                      2,
+
+                      // Use full parallelism, since at the time this sorter runs, it is the only sorter running.
+                      //
+                      // Typically, nothing else is running at all. Whenever there is more than one output partition,
+                      // the step prior to this localSort (typically hashPartition) should use buffered output channels
+                      // and therefore would have exited.
+                      //
+                      // In the case where there is one output partition, the step prior to this localSort may run
+                      // concurrently with the sorter, but in that case it will only have one output channel so won't
+                      // have many frames buffered.
+                      memoryParameters.getSuperSorterConcurrentProcessors(),
+                      memoryParameters.getSuperSorterMaxChannelsPerMerger(),
                       ShuffleSpec.UNLIMITED,
                       executionContext.cancellationId(),