| 1353 | } |
| 1354 | |
| 1355 | void addWindowSteps(QueryPlan & query_plan, |
| 1356 | const PlannerContextPtr & planner_context, |
| 1357 | WindowAnalysisResult & window_analysis_result, |
| 1358 | size_t max_step_description_length) |
| 1359 | { |
| 1360 | const auto & query_context = planner_context->getQueryContext(); |
| 1361 | const auto & settings = query_context->getSettingsRef(); |
| 1362 | |
| 1363 | auto & window_descriptions = window_analysis_result.window_descriptions; |
| 1364 | sortWindowDescriptions(window_descriptions); |
| 1365 | |
| 1366 | size_t window_descriptions_size = window_descriptions.size(); |
| 1367 | |
| 1368 | for (size_t i = 0; i < window_descriptions_size; ++i) |
| 1369 | { |
| 1370 | const auto & window_description = window_descriptions[i]; |
| 1371 | |
| 1372 | /** We don't need to sort again if the input from previous window already |
| 1373 | * has suitable sorting. Also don't create sort steps when there are no |
| 1374 | * columns to sort by, because the sort nodes are confused by this. It |
| 1375 | * happens in case of `over ()`. |
| 1376 | * Even if full_sort_description of both windows match, in case of different |
| 1377 | * partitioning we need to add a SortingStep to reshuffle data in the streams. |
| 1378 | */ |
| 1379 | |
| 1380 | bool need_sort = !window_description.full_sort_description.empty(); |
| 1381 | if (need_sort && i != 0) |
| 1382 | { |
| 1383 | const size_t effective_max_threads = getMaxThreadsForAvailableMemory( |
| 1384 | settings[Setting::max_threads], settings[Setting::max_threads_min_free_memory_per_thread]); |
| 1385 | need_sort = !sortDescriptionIsPrefix(window_description.full_sort_description, window_descriptions[i - 1].full_sort_description) |
| 1386 | || (effective_max_threads != 1 && window_description.partition_by.size() != window_descriptions[i - 1].partition_by.size()); |
| 1387 | } |
| 1388 | if (need_sort) |
| 1389 | { |
| 1390 | SortingStep::Settings sort_settings(query_context->getSettingsRef()); |
| 1391 | |
| 1392 | /// Window functions require fully sorted input. Applying sort_overflow_mode = 'break' |
| 1393 | /// would produce incomplete data and cause the pipeline to get stuck. |
| 1394 | sort_settings.size_limits.overflow_mode = OverflowMode::THROW; |
| 1395 | |
| 1396 | auto sorting_step = std::make_unique<SortingStep>( |
| 1397 | query_plan.getCurrentHeader(), |
| 1398 | window_description.full_sort_description, |
| 1399 | window_description.partition_by, |
| 1400 | 0 /*limit*/, |
| 1401 | sort_settings); |
| 1402 | sorting_step->setStepDescription("Sorting for window '" + window_description.window_name + "'", max_step_description_length); |
| 1403 | query_plan.addStep(std::move(sorting_step)); |
| 1404 | } |
| 1405 | |
| 1406 | // Fan out streams only for the last window to preserve the ordering between windows, |
| 1407 | // and WindowTransform works on single stream anyway. |
| 1408 | const bool streams_fan_out |
| 1409 | = settings[Setting::query_plan_enable_multithreading_after_window_functions] && ((i + 1) == window_descriptions_size); |
| 1410 | |
| 1411 | auto window_step |
| 1412 | = std::make_unique<WindowStep>(query_plan.getCurrentHeader(), window_description, window_description.window_functions, streams_fan_out); |
no test coverage detected