Method ensureOnHandler

server/src/main/java/io/crate/planner/Merge.java:65–105

Wrap the subPlan into a Merge plan if it isn't executed on the handler. @param projections projections to be applied on the subPlan or merge plan. These projections must not affect the limit, offset, order by or numOutputs If the subPlan contains a limit/offset or orderBy in i

(ExecutionPlan subExecutionPlan, PlannerContext plannerContext, List<Projection> projections)

     * add a {@link LimitAndOffset} AFTER the projections.
     */
    public static ExecutionPlan ensureOnHandler(ExecutionPlan subExecutionPlan, PlannerContext plannerContext, List<Projection> projections) {
        ResultDescription resultDescription = subExecutionPlan.resultDescription();
        assert resultDescription != null : "all plans must have a result description. Plan without: " +
                                           subExecutionPlan;

        // If a sub-Plan applies a limit it is usually limit+offset
        // So even if the execution is on the handler the limit may be too large and the final limit needs to be applied as well
        Projection limitAndOffset = ProjectionBuilder.limitAndOffsetOrEvalIfNeeded(
            resultDescription.limit(),
            resultDescription.offset(),
            resultDescription.numOutputs(),
            resultDescription.streamOutputs());
        if (ExecutionPhases.executesOnHandler(plannerContext.handlerNode(), resultDescription.nodeIds())) {
            return addProjections(subExecutionPlan, projections, resultDescription, limitAndOffset);
        }
        maybeUpdatePageSizeHint(subExecutionPlan, resultDescription.maxRowsPerNode());
        Collection<String> handlerNodeIds = Collections.singletonList(plannerContext.handlerNode());

        MergePhase mergePhase = new MergePhase(
            plannerContext.jobId(),
            plannerContext.nextExecutionPhaseId(),
            "mergeOnHandler",
            resultDescription.nodeIds().size(),
            1,
            handlerNodeIds,
            resultDescription.streamOutputs(),
            addProjection(projections, limitAndOffset),
            resultDescription.nodeIds(),
            DistributionInfo.DEFAULT_BROADCAST,
            resultDescription.orderBy()
        );
        return new Merge(
            subExecutionPlan,
            mergePhase,
            LimitAndOffset.NO_LIMIT,
            0,
            resultDescription.numOutputs(),
            resultDescription.limit(),
            resultDescription.orderBy()
        );
    }

    private static void maybeUpdatePageSizeHint(ExecutionPlan subExecutionPlan, int maxRowsPerNode) {
        if (Paging.shouldPage(maxRowsPerNode)) {