ClickHouse

1
#include <Processors/Merges/MergingSortedTransform.h>
2
#include <Processors/Transforms/ColumnGathererTransform.h>
3
#include <IO/WriteBuffer.h>
4

5
#include <Common/logger_useful.h>
6
#include <Common/formatReadable.h>
7

8
namespace DB
9
{
10

11
MergingSortedTransform::MergingSortedTransform(
12
    const Block & header,
13
    size_t num_inputs,
14
    const SortDescription & description_,
15
    size_t max_block_size_rows,
16
    size_t max_block_size_bytes,
17
    SortingQueueStrategy sorting_queue_strategy,
18
    UInt64 limit_,
19
    bool always_read_till_end_,
20
    WriteBuffer * out_row_sources_buf_,
21
    bool quiet_,
22
    bool use_average_block_sizes,
23
    bool have_all_inputs_)
24
    : IMergingTransform(
25
        num_inputs,
26
        header,
27
        header,
28
        have_all_inputs_,
29
        limit_,
30
        always_read_till_end_,
31
        header,
32
        num_inputs,
33
        description_,
34
        max_block_size_rows,
35
        max_block_size_bytes,
36
        sorting_queue_strategy,
37
        limit_,
38
        out_row_sources_buf_,
39
        use_average_block_sizes)
40
    , quiet(quiet_)
41
{
42
}
43

44
void MergingSortedTransform::onNewInput()
45
{
46
    algorithm.addInput();
47
}
48

49
void MergingSortedTransform::onFinish()
50
{
51
    if (quiet)
52
        return;
53

54
    const auto & merged_data = algorithm.getMergedData();
55

56
    auto log = getLogger("MergingSortedTransform");
57

58
    double seconds = total_stopwatch.elapsedSeconds();
59

60
    if (seconds == 0.0)
61
        LOG_DEBUG(log, "Merge sorted {} blocks, {} rows in 0 sec.", merged_data.totalChunks(), merged_data.totalMergedRows());
62
    else
63
        LOG_DEBUG(log, "Merge sorted {} blocks, {} rows in {} sec., {} rows/sec., {}/sec",
64
            merged_data.totalChunks(), merged_data.totalMergedRows(), seconds,
65
            merged_data.totalMergedRows() / seconds,
66
            ReadableSize(merged_data.totalAllocatedBytes() / seconds));
67
}
68

69
}
70