| 61 | } |
| 62 | |
| 63 | void WindowStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) |
| 64 | { |
| 65 | auto num_threads = pipeline.getNumThreads(); |
| 66 | |
| 67 | // This resize is needed for cases such as `over ()` when we don't have a |
| 68 | // sort node, and the input might have multiple streams. The sort node would |
| 69 | // have resized it. |
| 70 | if (window_description.full_sort_description.empty()) |
| 71 | pipeline.resize(1); |
| 72 | |
| 73 | pipeline.addSimpleTransform( |
| 74 | [&](const SharedHeader & /*header*/) |
| 75 | { |
| 76 | return std::make_shared<WindowTransform>( |
| 77 | input_headers.front(), output_header, window_description, window_functions); |
| 78 | }); |
| 79 | |
| 80 | if (streams_fan_out) |
| 81 | { |
| 82 | pipeline.resize(num_threads); |
| 83 | } |
| 84 | |
| 85 | assertBlocksHaveEqualStructure(pipeline.getHeader(), *output_header, |
| 86 | "WindowStep transform for '" + window_description.window_name + "'"); |
| 87 | } |
| 88 | |
| 89 | void WindowStep::describeActions(FormatSettings & settings) const |
| 90 | { |
nothing calls this directly
no test coverage detected