| 20 | { |
| 21 | |
| 22 | void GatherReceiveStep::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) |
| 23 | { |
| 24 | Pipes pipes; |
| 25 | |
| 26 | /// Read from all buckets |
| 27 | for (size_t i = 0; i < num_buckets; ++i) |
| 28 | { |
| 29 | pipes.push_back(Pipe(settings.exchange_lookup->createSource(output_header, ExchangeStreamId(exchange_id, i, 0)))); |
| 30 | } |
| 31 | |
| 32 | pipeline.init(Pipe::unitePipes(std::move(pipes))); |
| 33 | |
| 34 | if (maintain_sort_description && pipeline.getNumStreams() > 1) |
| 35 | { |
| 36 | pipeline.addTransform( |
| 37 | std::make_shared<MergingSortedTransform>( |
| 38 | output_header, |
| 39 | num_buckets, |
| 40 | *maintain_sort_description, |
| 41 | /* merge_block_size_rows */ DEFAULT_BLOCK_SIZE, |
| 42 | /* merge_block_size_bytes */ 0, |
| 43 | /* max_dynamic_subcolumns */ std::nullopt, |
| 44 | SortingQueueStrategy::Batch, |
| 45 | /* limit */ 0, |
| 46 | /* always_read_till_end */ false, |
| 47 | /* rows_sources_write_buf */ nullptr, |
| 48 | /* filter_column_name */ std::nullopt, |
| 49 | /* blocks_are_granules_size */ false)); |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | void GatherReceiveStep::serialize(Serialization & ctx) const |
| 54 | { |
nothing calls this directly
no test coverage detected