class StorageMaterializedView final : public IStorage, WithMutableContext
{
public:
    ...
private:
    /// Will be initialized in constructor
    StorageID target_table_id = StorageID::createEmpty();

    bool has_inner_table = false;
    ...
}

/** Storage. Describes the table. Responsible for
  * - storage of the table data;
  * - the definition in which files (or not in files) the data is stored;
  * - data lookups and appends;
  * - data storage structure (compression, etc.)
  * - concurrent access to data (locks, etc.)
  */

void StorageMaterializedView::read(
    ...)
{
    auto storage = getTargetTable();
    ...
    storage->read(query_plan, column_names, target_storage_snapshot, query_info, local_context, processed_stage, max_block_size, num_streams);
    ...
}

StoragePtr StorageMaterializedView::getTargetTable() const
{
    checkStackSize();
    return DatabaseCatalog::instance().getTable(target_table_id, getContext());
}

create table test( time DateTime) Engine=Memory();
create table source(time DateTime) Engine=Memory();
create materialized view mv_test to test as select time from source;
insert into table source values(now());

select * from test;

┌────────────────time─┐
│ 2023-02-25 18:51:41 │
└─────────────────────┘

select * from mv_test;

┌────────────────time─┐
│ 2023-02-25 18:51:41 │
└─────────────────────┘

explain  select * from test;

┌─explain───────────────────────────────────────────────────────────────────┐
│ Expression ((Projection + Before ORDER BY))                               │
│   SettingQuotaAndLimits (Set limits and quota after reading from storage) │
│     ReadFromStorage (Memory)                                              │
└───────────────────────────────────────────────────────────────────────────┘

explain  select * from mv_test;

┌─explain───────────────────────────────────────────────────────────────────┐
│ Expression ((Projection + Before ORDER BY))                               │
│   SettingQuotaAndLimits (Set limits and quota after reading from storage) │
│     SettingQuotaAndLimits (Lock destination table for MaterializedView)   │
│       ReadFromStorage (Memory)                                            │
└───────────────────────────────────────────────────────────────────────────┘

SinkToStoragePtr StorageMaterializedView::write(...)
{
    auto storage = getTargetTable();
    ...
    auto sink = storage->write(query, metadata_snapshot, local_context);
    ...
}

insert into table mv_test values(now());

select * from test;

┌────────────────time─┐
│ 2023-02-25 18:51:41 │
└─────────────────────┘
┌────────────────time─┐
│ 2023-02-25 19:11:20 │
└─────────────────────┘

StorageMaterializedView::StorageMaterializedView(
    const StorageID & table_id_,
    ContextPtr local_context,
    const ASTCreateQuery & query,
    const ColumnsDescription & columns_,
    bool attach_,
    const String & comment)
    : IStorage(table_id_), WithMutableContext(local_context->getGlobalContext())
{
    ...
    /// If the destination table is not set, use inner table
    has_inner_table = query.to_table_id.empty();  // Появились новые концепции inner table
    if (has_inner_table && !query.storage)  // Создание материализованного представлениячас,Илииметь ENGINE использовать inner таблицу или использовать TO использоватьвнешний стол
        throw Exception(
            "You must specify where to save results of a MaterializedView query: either ENGINE or an existing table in a TO clause",
            ErrorCodes::INCORRECT_QUERY);

    if (query.select->list_of_selects->children.size() != 1)
        throw Exception("UNION is not supported for MATERIALIZED VIEW", ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_MATERIALIZED_VIEW);

    ...

    // настраивать to_table_id
    if (!has_inner_table)
    {
        target_table_id = query.to_table_id;
    }
    else if (attach_)
    {
        /// If there is an ATTACH request, then the internal table must already be created.
        target_table_id = StorageID(getStorageID().database_name, generateInnerTableName(getStorageID()), query.to_inner_uuid);
    }
    else  // создать внутренний table
    {
        /// We will create a query to create an internal table.
        ...

        target_table_id = DatabaseCatalog::instance().getTable({manual_create_query->getDatabase(), manual_create_query->getTable()}, getContext())->getStorageID();  // Кажется ClickHouse иметьиндивидуальныйобщая ситуацияповерхностьзарегистрироватьсяповерхность    }
}

/** Interpreters interface for different queries.
  */
class IInterpreter
{
public:
    /** For queries that return a result (SELECT and similar), sets in BlockIO a stream from which you can read this result.
      * For queries that receive data (INSERT), sets a thread in BlockIO where you can write data.
      * For queries that do not require data and return nothing, BlockIO will be empty.
      */
    virtual BlockIO execute() = 0;
    ...
}

/** Interprets the INSERT query.
  */
class InterpreterInsertQuery : public IInterpreter, WithContext
{
public:
    ...
    /** Prepare a request for execution. Return block streams
      * - the stream into which you can write data to execute the query, if INSERT;
      * - the stream from which you can read the result of the query, if SELECT and similar;
      * Or nothing if the request INSERT SELECT (self-sufficient query - does not accept the input data, does not return the result).
      */
    BlockIO execute() override;
    ...
private
    ...
    Chain buildChainImpl(
        const StoragePtr & table,
        const StorageMetadataPtr & metadata_snapshot,
        const Block & query_sample_block,
        ThreadStatusesHolderPtr thread_status_holder,
        std::atomic_uint64_t * elapsed_counter_ms);
};

BlockIO InterpreterInsertQuery::execute()
{
    ...
    StoragePtr table = getTable(query);
    ...

    StoragePtr inner_table;
    if (const auto * mv = dynamic_cast<const StorageMaterializedView *>(table.get()))  // если insert query направлен на поверхность да StorageMaterializedView, целевая поверхность вынута и помещена inner_table Изменятьколичествосередина        inner_table = mv->getTargetTable();

    ...

    std::vector<Chain> out_chains;
    if (!distributed_pipeline || query.watch)
    {
        size_t out_streams_size = 1;

        if (query.select)  // иметь дело с INSERT ВЫБРАТЬ, игнорировать
        {
            ...
        }
        else if (query.watch)  // иметь дело с LIVE VIEW из WATCH заявление, просто проигнорируйте его
        {
            ...
        }

        for (size_t i = 0; i < out_streams_size; ++i)
        {
            auto out = buildChainImpl(table, metadata_snapshot, query_sample_block, nullptr, nullptr);  // Строить цепочка, важно! ! !
            out_chains.emplace_backmove(out);
        }
    }

    BlockIO res;

    /// What type of query: INSERT or INSERT SELECT or INSERT WATCH?
    if (distributed_pipeline)
    {
        res.pipeline = std::move(*distributed_pipeline);
    }
    else if (query.select || query.watch)
    {
        ...  // Просто игнорировать
    }
    else  // сосредоточиться наэтот филиал, запрос да INSERT час
    {
        res.pipeline = QueryPipelinemove(out_chains.at(0));  // Воля chain Первый созданный индивидуальный элемент возвращает BlockIO из pushing pipeline
        res.pipeline.setNumThreadsmin<size_t>(res.pipeline.getNumThreads(), settings.max_threads);  // настраивать query из конфигурации

        if (query.hasInlinedData() && !async_insert)
        {  // это да INSERT Предложение сопровождается VALUES (...) можно получить непосредственно из оператора середина, вставив изданные
            /// can execute without additional data
            auto pipe = getSourceFromASTInsertQuery(query_ptr, true, query_sample_block, getContext(), nullptr);
            res.pipeline.completemove(pipe);
        }
    }

    res.pipeline.addResourcesmove(resources);

    res.pipeline.addStorageHolder(table);  // Воля query из целевой поверхности вставить pipeline Ресурсная поверхность колонны
    if (inner_table)  // еслииметьматериализованный вид
        res.pipeline.addStorageHolder(inner_table);  // Пучокматериализованный также размещен видиз целевой поверхности pipeline из Ресурсная поверхность колонны

    return res;
}

Chain InterpreterInsertQuery::buildChainImpl(
    const StoragePtr & table,
    const StorageMetadataPtr & metadata_snapshot,
    const Block & query_sample_block,
    ThreadStatusesHolderPtr thread_status_holder,
    std::atomic_uint64_t * elapsed_counter_ms)
{
    ...

    /// We create a pipeline of several streams, into which we will write data.
    Chain out;

    /// Keep a reference to the context to make sure it stays alive until the chain is executed and destroyed
    out.addInterpreterContext(context_ptr);

    /// NOTE: we explicitly ignore bound materialized views when inserting into Kafka Storage.
    ///       Otherwise we'll get duplicates when MV reads same rows again from Kafka.
    if (table->noPushingToViews() && !no_destination)  // table->noPushingToViews() Используется для запрета материализованного вид вставить данные в KafkaEngine
    {
        auto sink = table->write(query_ptr, metadata_snapshot, context_ptr);
        sink->setRuntimeData(thread_status, elapsed_counter_ms);
        out.addSourcemove(sink);
    }
    else  // Строитьматериализованный видвставить pushingToViewChain, сосредоточься! ! !
    {
        out = buildPushingToViewsChain(table, metadata_snapshot, context_ptr, query_ptr, no_destination, thread_status_holder, elapsed_counter_ms);
    }

    ...

    return out;
}

Chain buildPushingToViewsChain(
    const StoragePtr & storage,
    const StorageMetadataPtr & metadata_snapshot,
    ContextPtr context,
    const ASTPtr & query_ptr,
    bool no_destination,
    ThreadStatusesHolderPtr thread_status_holder,
    std::atomic_uint64_t * elapsed_counter_ms,
    const Block & live_view_header)
{
    ...

    auto table_id = storage->getStorageID();
    Dependencies dependencies = DatabaseCatalog::instance().getDependencies(table_id);  // сосредоточиться, пройти table_id, получить "зависимость"этотповерхностьиз dependencies

    /// We need special context for materialized views insertions
    ContextMutablePtr select_context;
    ContextMutablePtr insert_context;
    ViewsDataPtr views_data;
    if (!dependencies.empty())
    {
        ...  // Разделяйте различные контексты запроса.
    }

    std::vector<Chain> chains;

    for (const auto & database_table : dependencies)
    {
        auto dependent_table = DatabaseCatalog::instance().getTable(database_table, context);
        auto dependent_metadata_snapshot = dependent_table->getInMemoryMetadataPtr();

        ASTPtr query;
        Chain out;

        ...

        if (auto * materialized_view = dynamic_cast<StorageMaterializedView *>(dependent_table.get()))  // Зависимостида MATERIALIZED VIEW
        {
            type = QueryViewsLogElement::ViewType::MATERIALIZED;
            result_chain.addTableLock(materialized_view->lockForShare(context->getInitialQueryId(), context->getSettingsRef().lock_acquire_timeout));

            StoragePtr inner_table = materialized_view->getTargetTable();  // получатьматериализованный видиз Цельповерхность            auto inner_table_id = inner_table->getStorageID();
            auto inner_metadata_snapshot = inner_table->getInMemoryMetadataPtr();
            query = dependent_metadata_snapshot->getSelectQuery().inner_query;
            target_name = inner_table_id.getFullTableName();

            /// Get list of columns we get from select query.
            auto header = InterpreterSelectQuery(query, select_context, SelectQueryOptions().analyze())
                .getSampleBlock();

            /// Insert only columns returned by select.
            Names insert_columns;
            const auto & inner_table_columns = inner_metadata_snapshot->getColumns();
            for (const auto & column : header)
            {
                /// But skip columns which storage doesn't have.
                if (inner_table_columns.hasPhysical(column.name))  // Уведомление,да совпадения по имени столбца,вместо позиции,этотсуществоватьиспользоватьматериализованный видчас Ошибаться легко
                    insert_columns.emplace_back(column.name);
            }

            InterpreterInsertQuery interpreter(nullptr, insert_context, false, false, false);  // Воляматериализованный видиз вставить логику также как InterpreterInsertQuery иметь дело с
            out = interpreter.buildChain(inner_table, inner_metadata_snapshot, insert_columns, thread_status_holder, view_counter_ms);
            out.addStorageHolder(dependent_table);
            out.addStorageHolder(inner_table);
        }
        else if (auto * live_view = dynamic_cast<StorageLiveView *>(dependent_table.get()))  // Зависимостида LIVE ПОСМОТРЕТЬ, игнорировать
        {
            ...
        }
        else if (auto * window_view = dynamic_cast<StorageWindowView *>(dependent_table.get()))  // Зависимостида WINDOW ПОСМОТРЕТЬ, игнорировать
        {
            ...
        }
        else
            out = buildPushingToViewsChain(
                dependent_table, dependent_metadata_snapshot, insert_context, ASTPtr(), false, thread_status_holder, view_counter_ms);  // Я так понимаю здесь дакаскадный материализованный вид филиала

        views_data->views.emplace_back(ViewRuntimeData{ //-V614
            std::move(query),
            out.getInputHeader(),
            database_table,
            nullptr,
            std::move(runtime_stats)});

        if (type == QueryViewsLogElement::ViewType::MATERIALIZED)
        {
            auto executing_inner_query = std::make_shared<ExecutingInnerQueryFromViewTransform>(
                storage_header, views_data->views.back(), views_data);
            executing_inner_query->setRuntimeData(view_thread_status, view_counter_ms);

            out.addSourcemove(executing_inner_query);
        }

        chains.emplace_backmove(out);

        /// Add the view to the query access info so it can appear in system.query_log
        if (!no_destination)
        {
            context->getQueryContext()->addQueryAccessInfo(
                backQuoteIfNeed(database_table.getDatabaseName()), views_data->views.back().runtime_stats->target_name, {}, "", database_table.getFullTableName());
        }
    }

    ...

    if (auto * live_view = dynamic_cast<StorageLiveView *>(storage.get()))
    {
        ...
    }
    else if (auto * window_view = dynamic_cast<StorageWindowView *>(storage.get()))
    {
        ...
    }
    /// Do not push to destination table if the flag is set
    else if (!no_destination)  // материализованный вид напиши логику
    {
        auto sink = storage->write(query_ptr, metadata_snapshot, context);  // Уведомление,Первыйиндивидуальныйпараметрдавходящийиз query_ptr,это даобъяснятьматериализованный видизданные также получаются непосредственно из запроса, без зависимости от поверхности
        metadata_snapshot->check(sink->getHeader().getColumnsWithTypeAndName());
        sink->setRuntimeData(thread_status, elapsed_counter_ms);
        result_chain.addSourcemove(sink);
    }

    ...

    return result_chain;
}

Dependencies DatabaseCatalog::getDependencies(const StorageID & from) const
{
    std::lock_guard lock{databases_mutex};
    auto iter = view_dependencies.find({from.getDatabaseName(), from.getTableName()});
    if (iter == view_dependencies.end())
        return {};
    return Dependencies(iter->second.begin(), iter->second.end());  // Найти из dependencies set Вставьте по порядку vector возвращаться
}

/// Table -> set of table-views that make SELECT from it.
using ViewDependencies = std::map<StorageID, std::set<StorageID>>;
using Dependencies = std::vector<StorageID>;
...
/// For some reason Context is required to get Storage from Database object
class DatabaseCatalog : boost::noncopyable, WithMutableContext
{
public:
    ...
private:
    ViewDependencies view_dependencies TSA_GUARDED_BY(databases_mutex);
}

struct StorageID
{
    String database_name;
    String table_name;
    UUID uuid = UUIDHelpers::Nil;
    ...
private:
    ...
};

CREATE MATERIALIZED VIEW [IF NOT EXISTS] [db.]table_name [ON CLUSTER] [TO[db.]name] [ENGINE = engine] [POPULATE] AS SELECT ...

CREATE MATERIALIZED VIEW mv (
    a Int64,
    d Date,
    cnt Int64
) ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(d)
ORDER BY (a,d)
POPULATE
AS
SELECT
    a,
    d,count() AS cnt -- должно быть Уведомление AS cnt
 ИЗ источника GROUP BY a, d;

select * from system.text_log where logger_name like '%Kafka%'

flowchart LR
    A[kafka_engine] -->|when !error| B[kafka_data]
    A[kafka_engine] -->|when error| C[kafka_errors]

CREATE TABLE default.kafka_engine
(
    `i` Int64,
    `s` String
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092'
kafka_topic_list = 'topic',
kafka_group_name = 'clickhouse',
kafka_format = 'JSONEachRow',
kafka_handle_error_mode='stream';

CREATE MATERIALIZED VIEW default.kafka_data
(
    `i` Int64,
    `s` String
)
ENGINE = MergeTree
ORDER BY (`i`)
AS
SELECT
    `i`,
    `s`
FROM default.kafka_engine
WHERE length(_error) = 0

CREATE MATERIALIZED VIEW default.kafka_errors
(
    `topic` String,
    `partition` Int64,
    `offset` Int64,
    `raw` String,
    `error` String
)
ENGINE = MergeTree
ORDER BY (topic, partition, offset)
SETTINGS index_granularity = 8192 AS
SELECT
    _topic AS topic,
    _partition AS partition,
    _offset AS offset,
    _raw_message AS raw,
    _error AS error
FROM default.kafka_engine
WHERE length(_error) > 0

CREATE TABLE IF NOT EXISTS entry (
    json_str String
)ENGINE = Null;

CREATE TABLE IF NOT EXISTS dest (
    two_diemnsional_array Array(Array(String))
)ENGINE = MergeTree()
ORDER BY tuple();

CREATE MATERIALIZED VIEW mv_dest TO dest
AS
SELECT
    JSONExtract(json_str, 'Array(Array(String))') as two_diemnsional_array
FROM entry;

CREATE TABLE IF NOT EXISTS event  -- оригинальныйповерхность
(    `app_id` LowCardinality(String) CODEC(ZSTD(9)),
    `time` Int64 CODEC(Delta, ZSTD(9)),
    `user_id` String CODEC(Delta, ZSTD(9)),
    `device_id` String CODEC(Delta, ZSTD(9)),
    `d1` String CODEC(ZSTD(9)),
    `d2` String CODEC(ZSTD(9)),
    `d3` String CODEC(ZSTD(9)),
    `d4` String CODEC(ZSTD(9)),
    `d5` String CODEC(ZSTD(9)),
    `d6` String CODEC(ZSTD(9)),
    `v1` Int64 CODEC(T64, ZSTD(9)),
    `v2` Int64 CODEC(T64, ZSTD(9)),
    `v3` Int64 CODEC(T64, ZSTD(9)),
    `v4` Int64 CODEC(T64, ZSTD(9)),
    `v5` Int64 CODEC(T64, ZSTD(9)),
    `v6` Int64 CODEC(T64, ZSTD(9))
)ENGINE = MergeTree()
PARTITION BY intDiv(time, 2592000000)
ORDER BY (app_id, time)
TTL toDate(intDiv(time, 1000)) + toIntervalMonth(1)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1, use_minimalistic_part_header_in_zookeeper = 1;

CREATE TABLE IF NOT EXISTS event_agg_5min
(
    `app_id` LowCardinality(String) CODEC(ZSTD(9)),
    `time` Int64 CODEC(Delta, ZSTD(9)),
    `user_cnt` AggregateFunction(uniq, String) CODEC(ZSTD(9)),
    `device_cnt` AggregateFunction(uniq, String) CODEC(ZSTD(9)),
    `cnt` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `d1` String CODEC(ZSTD(9)),
    `d2` String CODEC(ZSTD(9)),
    `d3` String CODEC(ZSTD(9)),
    `d4` String CODEC(ZSTD(9)),
    `d5` String CODEC(ZSTD(9)),
    `d6` String CODEC(ZSTD(9)),
    `v1` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `v2` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `v3` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `v4` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `v5` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9)),
    `v6` SimpleAggregateFunction(sum, Int64) CODEC(T64, ZSTD(9))
)
ENGINE = AggregatingMergeTree()  -- Инкрементная предварительная агрегация после размещения
PARTITION BY intDiv(time, 2592000000)
PRIMARY KEY (app_id, time)
ORDER BY (app_id, time, d1, d2, d3, d4, d5, d6)  -- ORDER BY Необходимо включить все столбцы «Ярлык»
TTL toDate(intDiv(time, 1000)) + toIntervalMonth(1)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1, use_minimalistic_part_header_in_zookeeper = 1;

CREATE MATERIALIZED VIEW IF NOT EXISTS mv_event_agg_5min TO event_agg_5min AS
SELECT
    app_id,
    intDiv(entrance_time, 300000) * 300000 AS entrance_time,  -- 5 Агрегация минут
    uniqState(user_id) AS user_cnt,  -- AggregateFunctionиспользовать -State полимеризация
    uniqState(device_id) AS device_cnt,  -- Уведомление AS Переименовать
    count(*) AS cnt,
    d1,
    d2,
    d3,
    d4,
    d5,
    d6,
    sum(v1) as v1,  -- SimpleAggregateFunction прямойиспользоватьполимеризацияфункция
    sum(v2) as v2,
    sum(v3) as v3,
    sum(v4) as v4,
    sum(v5) as v5,
    sum(v6) as v6
FROM event
GROUP BY  -- одининдивидуальный Block Внутрипредварительнополимеризация
    app_id,
    category,
    d1,
    d2,
    d3,
    d4,
    d5,
    d6;