Method execute

src/DataStreams/TTLAggregationAlgorithm.cpp:82–186

}

void TTLAggregationAlgorithm::execute(Block & block)
{

    bool some_rows_were_aggregated = false;
    MutableColumns result_columns = header.cloneEmptyColumns();

    if (!block) /// Empty block -- no more data, but we may still have some accumulated rows
    {
        if (!aggregation_result.empty()) /// Still have some aggregated data, let's update TTL
        {
            finalizeAggregates(result_columns);
            some_rows_were_aggregated = true;
        }
        else /// No block, all aggregated, just finish
        {
            return;
        }
    }
    else
    {
        const auto & column_names = header.getNames();
        MutableColumns aggregate_columns = header.cloneEmptyColumns();

        auto ttl_column = executeExpressionAndGetColumn(description.expression, block, description.result_column);
        auto where_column = executeExpressionAndGetColumn(description.where_expression, block, description.where_result_column);

        size_t rows_aggregated = 0;
        size_t current_key_start = 0;
        size_t rows_with_current_key = 0;

        for (size_t i = 0; i < block.rows(); ++i)
        {
            UInt32 cur_ttl = getTimestampByIndex(ttl_column.get(), i);
            bool where_filter_passed = !where_column || where_column->getBool(i);
            bool ttl_expired = isTTLExpired(cur_ttl) && where_filter_passed;

            bool same_as_current = true;
            for (size_t j = 0; j < description.group_by_keys.size(); ++j)
            {
                const String & key_column = description.group_by_keys[j];
                const IColumn * values_column = block.getByName(key_column).column.get();
                if (!same_as_current || (*values_column)[i] != current_key_value[j])
                {
                    values_column->get(i, current_key_value[j]);
                    same_as_current = false;
                }
            }

            if (!same_as_current)
            {
                if (rows_with_current_key)
                {
                    some_rows_were_aggregated = true;
                    calculateAggregates(aggregate_columns, current_key_start, rows_with_current_key);
                }
                finalizeAggregates(result_columns);

                current_key_start = rows_aggregated;