ClickHouse

1
#include <Processors/QueryPlan/ReadFromStreamLikeEngine.h>
2

3
#include <Interpreters/InterpreterSelectQuery.h>
4
#include <QueryPipeline/QueryPipelineBuilder.h>
5

6
namespace DB
7
{
8

9
namespace ErrorCodes
10
{
11
extern const int QUERY_NOT_ALLOWED;
12
}
13

14
ReadFromStreamLikeEngine::ReadFromStreamLikeEngine(
15
    const Names & column_names_,
16
    const StorageSnapshotPtr & storage_snapshot_,
17
    std::shared_ptr<const StorageLimitsList> storage_limits_,
18
    ContextPtr context_)
19
    : ISourceStep{DataStream{.header = storage_snapshot_->getSampleBlockForColumns(column_names_)}}
20
    , WithContext{context_}
21
    , storage_limits{std::move(storage_limits_)}
22
{
23
}
24

25
void ReadFromStreamLikeEngine::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
26
{
27
    if (!getContext()->getSettingsRef().stream_like_engine_allow_direct_select)
28
        throw Exception(
29
            ErrorCodes::QUERY_NOT_ALLOWED, "Direct select is not allowed. To enable use setting `stream_like_engine_allow_direct_select`");
30

31
    auto pipe = makePipe();
32

33
    /// Add storage limits.
34
    for (const auto & processor : pipe.getProcessors())
35
        processor->setStorageLimits(storage_limits);
36

37
    /// Add to processors to get processor info through explain pipeline statement.
38
    for (const auto & processor : pipe.getProcessors())
39
        processors.emplace_back(processor);
40

41
    pipeline.init(std::move(pipe));
42
}
43
}
44