ClickHouse

1
#include <Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.h>
2
#include <Processors/Merges/Algorithms/MergeTreePartLevelInfo.h>
3

4

5
namespace DB
6
{
7

8
IMergingAlgorithmWithDelayedChunk::IMergingAlgorithmWithDelayedChunk(Block header_, size_t num_inputs, SortDescription description_)
9
    : description(std::move(description_))
10
    , header(std::move(header_))
11
    , current_inputs(num_inputs)
12
    , cursors(num_inputs)
13
    , inputs_origin_merge_tree_part_level(num_inputs)
14
{
15
}
16

17
void IMergingAlgorithmWithDelayedChunk::initializeQueue(Inputs inputs)
18
{
19
    current_inputs = std::move(inputs);
20

21
    for (size_t source_num = 0; source_num < current_inputs.size(); ++source_num)
22
    {
23
        if (!current_inputs[source_num].chunk)
24
            continue;
25

26
        cursors[source_num] = SortCursorImpl(
27
            header, current_inputs[source_num].chunk.getColumns(), description, source_num, current_inputs[source_num].permutation);
28

29
        inputs_origin_merge_tree_part_level[source_num] = getPartLevelFromChunk(current_inputs[source_num].chunk);
30
    }
31

32
    queue = SortingQueue<SortCursor>(cursors);
33
}
34

35
void IMergingAlgorithmWithDelayedChunk::updateCursor(Input & input, size_t source_num)
36
{
37
    auto & current_input = current_inputs[source_num];
38

39
    /// Extend lifetime of last chunk.
40
    last_chunk.swap(current_input.chunk);
41
    last_chunk_sort_columns = std::move(cursors[source_num].sort_columns);
42

43
    current_input.swap(input);
44
    cursors[source_num].reset(current_input.chunk.getColumns(), header, current_input.permutation);
45

46
    inputs_origin_merge_tree_part_level[source_num] = getPartLevelFromChunk(current_input.chunk);
47

48
    queue.push(cursors[source_num]);
49
}
50

51
}
52