Support different concurrency settings for table writer
If TableWriter input is fixed distributed (e.g.: aggregation or join)
the actual aggregation or join concurrency will be set to the
table writer concurrency
@@ -508,6 +508,9 @@ public class AddLocalExchanges
PlanWithProperties tableWriter;
if (!originalTableWriterNode.getTablePartitioningScheme().isPresent()) {
+ int taskWriterCount = getTaskWriterCount(session);
+ int taskConcurrency = getTaskConcurrency(session);
+ if (taskWriterCount == taskConcurrency) {
tableWriter = planAndEnforceChildren(
new TableWriterNode(
originalTableWriterNode.getId(),
@@ -524,6 +527,27 @@ public class AddLocalExchanges
fixedParallelism(),
fixedParallelism());
}
+ else {
+ PlanWithProperties source = originalTableWriterNode.getSource().accept(this, defaultParallelism(session));
+ PlanWithProperties exchange = deriveProperties(
+ roundRobinExchange(idAllocator.getNextId(), LOCAL, source.getNode()),
+ source.getProperties());
+ tableWriter = deriveProperties(
+ new TableWriterNode(
+ originalTableWriterNode.getId(),
+ exchange.getNode(),
+ originalTableWriterNode.getTarget(),
+ variableAllocator.newVariable("partialrowcount", BIGINT),
+ variableAllocator.newVariable("partialfragments", VARBINARY),
+ variableAllocator.newVariable("partialcontext", VARBINARY),
+ originalTableWriterNode.getColumns(),
+ originalTableWriterNode.getColumnNames(),
+ originalTableWriterNode.getTablePartitioningScheme(),
+ originalTableWriterNode.getPreferredShufflePartitioningScheme(),
+ statisticAggregations.map(StatisticAggregations.Parts::getPartialAggregation)),
+ exchange.getProperties());
+ }
+ }
else {
PlanWithProperties source = originalTableWriterNode.getSource().accept(this, fixedParallelism());
PlanWithProperties exchange = deriveProperties(
