| 524 | } |
| 525 | |
| 526 | void SortingStep::fullSort( |
| 527 | QueryPipelineBuilder & pipeline, const SortDescription & result_sort_desc, const UInt64 limit_, const bool skip_partial_sort) |
| 528 | { |
| 529 | scatterByPartitionIfNeeded(pipeline); |
| 530 | |
| 531 | fullSortStreams(pipeline, sort_settings, result_sort_desc, limit_, skip_partial_sort, threshold_tracker); |
| 532 | |
| 533 | addPerStreamLimitByIfNeeded(pipeline, result_sort_desc); |
| 534 | |
| 535 | /// If there are several streams, then we merge them into one |
| 536 | if (pipeline.getNumStreams() > 1 && (partition_by_description.empty() || pipeline.getNumThreads() == 1)) |
| 537 | { |
| 538 | auto transform = std::make_shared<MergingSortedTransform>( |
| 539 | pipeline.getSharedHeader(), |
| 540 | pipeline.getNumStreams(), |
| 541 | result_sort_desc, |
| 542 | sort_settings.max_block_size, |
| 543 | /*max_block_size_bytes=*/0, |
| 544 | /*max_dynamic_subcolumns*/std::nullopt, |
| 545 | SortingQueueStrategy::Batch, |
| 546 | limit_, |
| 547 | always_read_till_end); |
| 548 | |
| 549 | pipeline.addTransform(std::move(transform)); |
| 550 | } |
| 551 | else if (apply_virtual_row_conversions) |
| 552 | { |
| 553 | pipeline.addSimpleTransform(RemoveVirtualRowTransform::create); |
| 554 | } |
| 555 | } |
| 556 | |
| 557 | void SortingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) |
| 558 | { |
nothing calls this directly
no test coverage detected