```
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] [                                                                                                                                                                                                                                        <=>               ]
```

{
    auto creator_fn = [](const StorageFactory::Arguments & args)
    {
        ASTs & engine_args = args.engine_args;
        size_t args_count = engine_args.size();
        bool has_settings = args.storage_def->settings;
        auto kafka_settings = std::make_unique<KafkaSettings>();
        auto named_collection = getExternalDataSourceConfiguration(args.engine_args, *kafka_settings, args.getLocalContext());
        if (has_settings)
        {
            kafka_settings->loadFromQuery(*args.storage_def);
        }
        
        #define CHECK_KAFKA_STORAGE_ARGUMENT(ARG_NUM, PAR_NAME, EVAL)       \
            /* One of the four required arguments is not specified */       \
            if (args_count < (ARG_NUM) && (ARG_NUM) <= 4 &&                 \
                !kafka_settings->PAR_NAME.changed)                          \
            {                                                               \
                throw Exception(                                            \
                    "Required parameter '" #PAR_NAME "' "                   \
                    "for storage Kafka not specified",                      \