| 486 | } |
| 487 | |
| 488 | Pipe ReadFromMergeTree::read( |
| 489 | RangesInDataParts parts_with_range, |
| 490 | Names required_columns, |
| 491 | ReadType read_type, |
| 492 | size_t max_streams, |
| 493 | size_t min_marks_for_concurrent_read, |
| 494 | bool use_uncompressed_cache, |
| 495 | const std::shared_ptr<DelayedSkipIndex> & delayed_index) |
| 496 | { |
| 497 | if (read_type == ReadType::Default && max_streams > 1) |
| 498 | { |
| 499 | if (unlikely(delayed_index != nullptr)) |
| 500 | { |
| 501 | throw Exception(ErrorCodes::LOGICAL_ERROR, "Some skip index is delayed to pipeline execution stage"); |
| 502 | } |
| 503 | return readFromPool(parts_with_range, required_columns, max_streams, min_marks_for_concurrent_read, use_uncompressed_cache); |
| 504 | } |
| 505 | |
| 506 | auto pipe = readInOrder(parts_with_range, required_columns, |
| 507 | read_type, use_uncompressed_cache, delayed_index); |
| 508 | |
| 509 | /// Use ConcatProcessor to concat sources together. |
| 510 | /// It is needed to read in parts order (and so in PK order) if single thread is used. |
| 511 | if (read_type == ReadType::Default && pipe.numOutputPorts() > 1) |
| 512 | pipe.addTransform(std::make_shared<ConcatProcessor>(pipe.getHeader(), pipe.numOutputPorts())); |
| 513 | |
| 514 | return pipe; |
| 515 | } |
| 516 | |
| 517 | namespace |
| 518 | { |
no test coverage detected