| 52 | } |
| 53 | |
| 54 | void BlocksMarshallingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) |
| 55 | { |
| 56 | // The getNumStreams() == 1 is special, because it may indicate that pipeline ended with a sorting or aggregation, i.e. we should preserve chunks order. |
| 57 | const bool single_stream = pipeline.getNumStreams() == 1; |
| 58 | if (single_stream) |
| 59 | pipeline.addTransform(std::make_shared<AddSequenceNumber>(pipeline.getSharedHeader())); |
| 60 | const size_t num_threads = pipeline.getNumThreads(); |
| 61 | pipeline.resize(num_threads); |
| 62 | pipeline.addSimpleTransform([&](const SharedHeader & header, Pipe::StreamType stream_type) -> ProcessorPtr |
| 63 | { |
| 64 | /// Skip marshalling for totals and extremes streams because `IOutputFormat::prepareTotals` |
| 65 | /// may call `cut` on columns, which `ColumnBLOB` does not support. |
| 66 | if (stream_type != Pipe::StreamType::Main) |
| 67 | return nullptr; |
| 68 | return std::make_shared<MarshallBlocksTransform>(header, settings.block_marshalling_callback); |
| 69 | }); |
| 70 | if (single_stream) |
| 71 | pipeline.addTransform(std::make_shared<SortChunksBySequenceNumber>(pipeline.getHeader(), num_threads)); |
| 72 | } |
| 73 | |
| 74 | QueryPlanStepPtr BlocksMarshallingStep::deserialize(Deserialization & ctx) |
| 75 | { |
nothing calls this directly
no test coverage detected