Add check that dynamic filtering is not enabled with grouped execution
Dynamic filtering assumes that join build side is evaulated fixed
(task_concurrency) amount of times. This constraint is not satisfied
when grouped execution is enabled.
cherry-pick of:
@@ -2179,7 +2179,7 @@ public class LocalExecutionPlanner
ImmutableList.Builder<OperatorFactory> factoriesBuilder = new ImmutableList.Builder<>();
factoriesBuilder.addAll(buildSource.getOperatorFactories());
- createDynamicFilter(node, context, partitionCount).ifPresent(
+ createDynamicFilter(buildSource, node, context, partitionCount).ifPresent(
filter -> factoriesBuilder.add(createDynamicFilterSourceOperatorFactory(filter, node.getId(), buildSource, buildContext)));
HashBuilderOperatorFactory hashBuilderOperatorFactory = new HashBuilderOperatorFactory(
@@ -2233,7 +2233,7 @@ public class LocalExecutionPlanner
getDynamicFilteringMaxPerDriverSize(context.getSession()));
}
- private Optional<LocalDynamicFilter> createDynamicFilter(AbstractJoinNode node, LocalExecutionPlanContext context, int partitionCount)
+ private Optional<LocalDynamicFilter> createDynamicFilter(PhysicalOperation buildSource, AbstractJoinNode node, LocalExecutionPlanContext context, int partitionCount)
{
if (!isEnableDynamicFiltering(context.getSession())) {
return Optional.empty();
@@ -2241,6 +2241,9 @@ public class LocalExecutionPlanner
if (node.getDynamicFilters().isEmpty()) {
return Optional.empty();
}
+ checkState(
+ buildSource.getPipelineExecutionStrategy() != GROUPED_EXECUTION,
+ "Dynamic filtering cannot be used with grouped execution");
LocalDynamicFiltersCollector collector = context.getDynamicFiltersCollector();
return LocalDynamicFilter
.create(node, partitionCount)
@@ -2355,7 +2358,7 @@ public class LocalExecutionPlanner
// add collector to the source
int partitionCount = buildContext.getDriverInstanceCount().orElse(1);
- Optional<LocalDynamicFilter> localDynamicFilter = createDynamicFilter(node, context, partitionCount);
+ Optional<LocalDynamicFilter> localDynamicFilter = createDynamicFilter(buildSource, node, context, partitionCount);
localDynamicFilter.ifPresent(filter -> factoriesBuilder.add(createDynamicFilterSourceOperatorFactory(filter, node.getId(), buildSource, buildContext)));
factoriesBuilder.add(setBuilderOperatorFactory);
