Method writeIntoWindowView

src/Storages/WindowView/StorageWindowView.cpp:1498–1699

}

void StorageWindowView::writeIntoWindowView(
    StorageWindowView & window_view, Block && block, Chunk::ChunkInfoCollection && chunk_infos, ContextPtr local_context)
{
    window_view.throwIfWindowViewIsDisabled(local_context);
    while (window_view.modifying_query)
        std::this_thread::sleep_for(std::chrono::milliseconds(100));

    const size_t block_rows = block.rows();
    if (!window_view.is_proctime && window_view.max_watermark == 0 && block_rows > 0)
    {
        std::lock_guard lock(window_view.fire_signal_mutex);
        const auto & window_column = block.getByName(window_view.timestamp_column_name);
        const ColumnUInt32::Container & window_end_data = static_cast<const ColumnUInt32 &>(*window_column.column).getData();
        UInt32 first_record_timestamp = window_end_data[0];
        window_view.max_watermark = window_view.getWindowUpperBound(first_record_timestamp);

        LOG_TRACE(window_view.log, "New max watermark: {}", window_view.max_watermark);
    }

    Pipe pipe(std::make_shared<SourceFromSingleChunk>(std::make_shared<const Block>(block)));

    UInt32 lateness_bound = 0;
    UInt32 t_max_watermark = 0;
    UInt32 t_max_timestamp = 0;
    UInt32 t_max_fired_watermark = 0;
    {
        std::lock_guard lock(window_view.fire_signal_mutex);
        t_max_fired_watermark = window_view.max_fired_watermark;
        t_max_watermark = window_view.max_watermark;
        t_max_timestamp = window_view.max_timestamp;
    }

    // Filter outdated data
    if (window_view.allowed_lateness && t_max_timestamp != 0)
    {
        lateness_bound = addTime(t_max_timestamp, window_view.lateness_kind, -window_view.lateness_num_units, *window_view.time_zone);

        if (window_view.is_watermark_bounded)
        {
            UInt32 watermark_lower_bound
                = addTime(t_max_watermark, window_view.slide_kind, -window_view.slide_num_units, *window_view.time_zone);

            lateness_bound = std::min(watermark_lower_bound, lateness_bound);
        }
    }
    else if (!window_view.is_time_column_func_now)
    {
        lateness_bound = t_max_fired_watermark;
    }

    if (lateness_bound > 0) /// Add filter, which leaves rows with timestamp >= lateness_bound
    {
        auto filter_function = makeASTOperator(
            "greaterOrEquals",
            make_intrusive<ASTIdentifier>(window_view.timestamp_column_name),
            make_intrusive<ASTLiteral>(lateness_bound));

        ASTPtr query = filter_function;