```
ubuntu@ip-172-31-79-147:/opt/downloads$ clickhouse-local --max_threads 8 --query "WITH parseDateTimeBestEffort(extract(_path, 'pageviews-([\\d\\-]+)\\.gz$')) AS time, splitByChar(' ', line) AS values, splitByChar('.', values[1]) AS projects SELECT time, projects[1] AS project, projects[2] AS subproject, values[2] AS path, (values[3])::UInt64 AS hits FROM file('pageviews*.gz', LineAsString) WHERE length(values) >= 3" | pv > /dev/null
^C01GiB 0:01:08 [ 129MiB/s] [                                                                                                                                                                   <=>                                                                                    ]

ubuntu@ip-172-31-79-147:/opt/downloads$ clickhouse-local --max_threads 8 --query "WITH parseDateTimeBestEffort(extract(toLowCardinality(_path), 'pageviews-([\\d\\-]+)\\.gz$')) AS time, splitByChar(' ', line) AS values, splitByChar('.', values[1]) AS projects SELECT time, projects[1] AS project, projects[2] AS subproject, values[2] AS path, (values[3])::UInt64 AS hits FROM file('pageviews*.gz', LineAsString) WHERE length(values) >= 3" | pv > /dev/null
^C46GiB 0:00:48 [ 175MiB/s] [                                                                                                                                                                                                                                        <=>               ]
```

{
        StorageKafka * self = reinterpret_cast<StorageKafka *>(ctx);
        std::lock_guard lock(self->thread_statuses_mutex);
        const auto it = std::find_if(self->thread_statuses.begin(), self->thread_statuses.end(), [](const auto & thread_status_ptr)
        {
            return thread_status_ptr.get() == current_thread;
        });
        if (it == self->thread_statuses.end())
            throw Exception("No thread status for this librdkafka thread.", ErrorCodes::LOGICAL_ERROR);
        self->thread_statuses.erase(it);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
    }
    static rd_kafka_resp_err_t rdKafkaOnNew(rd_kafka_t * rk, const rd_kafka_conf_t *, void * ctx, char * , size_t )
    {
        StorageKafka * self = reinterpret_cast<StorageKafka *>(ctx);
        rd_kafka_resp_err_t status;
        status = rd_kafka_interceptor_add_on_thread_start(rk, "init-thread", rdKafkaOnThreadStart, ctx);
        if (status != RD_KAFKA_RESP_ERR_NO_ERROR)
        {
            LOG_ERROR(self->log, "Cannot set on thread start interceptor due to {} error", status);
            return status;
        }
        status = rd_kafka_interceptor_add_on_thread_exit(rk, "exit-thread", rdKafkaOnThreadExit, ctx);
        if (status != RD_KAFKA_RESP_ERR_NO_ERROR)
            LOG_ERROR(self->log, "Cannot set on thread exit interceptor due to {} error", status);
        return status;
    }
    static rd_kafka_resp_err_t rdKafkaOnConfDup(rd_kafka_conf_t * new_conf, const rd_kafka_conf_t * , size_t , const char ** , void * ctx)