Preserve order for root for SINGLE distribution only

We need to set preserveOrderWithinPartition flag only if root stage of
plan requires total ordering. The need of total ordering implies that
SINGLE distribution will be used for root stage.

Note there still can be cases when SINGLE distribution is used but we do
not really need to preserve order. Improving on that is a followup work
as it requires annotating plan with information if output ordering is
required or not.

core/trino-main/src/main/java/io/trino/execution/scheduler/EventDrivenFaultTolerantQueryScheduler.java

@@ -144,6 +144,7 @@
 import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.trino.spi.StandardErrorCode.REMOTE_HOST_GONE;
 import static io.trino.sql.planner.SystemPartitioningHandle.COORDINATOR_DISTRIBUTION;
+import static io.trino.sql.planner.SystemPartitioningHandle.SINGLE_DISTRIBUTION;
 import static io.trino.util.Failures.toFailure;
 import static java.lang.Math.max;
 import static java.lang.Math.min;
@@ -798,10 +799,12 @@ private void createStageExecution(SubPlan subPlan, boolean rootFragment, int sch
 
                 FaultTolerantPartitioningScheme sinkPartitioningScheme = partitioningSchemeFactory.get(fragment.getPartitioningScheme().getPartitioning().getHandle());
                 ExchangeContext exchangeContext = new ExchangeContext(queryStateMachine.getQueryId(), new ExchangeId("external-exchange-" + stage.getStageId().getId()));
+
+                boolean preserveOrderWithinPartition = rootFragment && stage.getFragment().getPartitioning().equals(SINGLE_DISTRIBUTION);
                 Exchange exchange = closer.register(exchangeManager.createExchange(
                         exchangeContext,
                         sinkPartitioningScheme.getPartitionCount(),
-                        rootFragment));
+                        preserveOrderWithinPartition));
 
                 boolean coordinatorStage = stage.getFragment().getPartitioning().equals(COORDINATOR_DISTRIBUTION);