(DependencyCarrier executor,
PlannerContext plannerContext,
Set<PlanHint> planHints,
ProjectionBuilder projectionBuilder,
int limit,
int offset,
@Nullable OrderBy order,
@Nullable Integer pageSizeHint,
Row params,
SubQueryResults subQueryResults)
| 139 | } |
| 140 | |
| 141 | @SuppressWarnings("unchecked") |
| 142 | @Override |
| 143 | public ExecutionPlan build(DependencyCarrier executor, |
| 144 | PlannerContext plannerContext, |
| 145 | Set<PlanHint> planHints, |
| 146 | ProjectionBuilder projectionBuilder, |
| 147 | int limit, |
| 148 | int offset, |
| 149 | @Nullable OrderBy order, |
| 150 | @Nullable Integer pageSizeHint, |
| 151 | Row params, |
| 152 | SubQueryResults subQueryResults) { |
| 153 | InputColumns.SourceSymbols sourceSymbols = new InputColumns.SourceSymbols(source.outputs()); |
| 154 | |
| 155 | SubQueryAndParamBinder binder = new SubQueryAndParamBinder(params, subQueryResults); |
| 156 | Function<Symbol, Symbol> toInputCols = binder.andThen(s -> InputColumns.create(s, sourceSymbols)); |
| 157 | |
| 158 | List<WindowFunction> boundWindowFunctions = (List<WindowFunction>)(List<?>) Lists.map(windowFunctions, toInputCols); |
| 159 | List<Projection> projections = new ArrayList<>(); |
| 160 | WindowAggProjection windowAggProjection = new WindowAggProjection( |
| 161 | windowDefinition.map(toInputCols), |
| 162 | boundWindowFunctions, |
| 163 | InputColumns.create(this.standalone, sourceSymbols) |
| 164 | ); |
| 165 | projections.add(windowAggProjection); |
| 166 | ExecutionPlan sourcePlan = source.build( |
| 167 | executor, |
| 168 | plannerContext, |
| 169 | planHints, |
| 170 | projectionBuilder, |
| 171 | LimitAndOffset.NO_LIMIT, |
| 172 | LimitAndOffset.NO_OFFSET, |
| 173 | null, |
| 174 | pageSizeHint, |
| 175 | params, |
| 176 | subQueryResults |
| 177 | ); |
| 178 | ResultDescription resultDescription = sourcePlan.resultDescription(); |
| 179 | boolean executesOnHandler = executesOnHandler(plannerContext.handlerNode(), resultDescription.nodeIds()); |
| 180 | boolean nonDistExecution = windowDefinition.partitions().isEmpty() |
| 181 | || resultDescription.hasRemainingLimitOrOffset() |
| 182 | || executesOnHandler; |
| 183 | if (nonDistExecution) { |
| 184 | sourcePlan = Merge.ensureOnHandler(sourcePlan, plannerContext); |
| 185 | for (Projection projection : projections) { |
| 186 | sourcePlan.addProjection(projection); |
| 187 | } |
| 188 | return sourcePlan; |
| 189 | } |
| 190 | |
| 191 | Symbol firstPartition = windowDefinition.partitions().getFirst(); |
| 192 | int index = source.outputs().indexOf(firstPartition); |
| 193 | if (index == -1) { |
| 194 | // PARTITION BY on subscript/expression not contained in source outputs |
| 195 | // Modulo bucketing in nodes < 6.1.1 can't handle arbitrary expressions -> fallback to non distributed execution |
| 196 | if (plannerContext.clusterState().nodes().getSmallestNonClientNodeVersion().before(Version.V_6_1_1)) { |
| 197 | sourcePlan = Merge.ensureOnHandler(sourcePlan, plannerContext); |
| 198 | for (Projection projection : projections) { |
nothing calls this directly
no test coverage detected